diff --git a/.dockerignore b/.dockerignore index bbf27a46..9a102e73 100644 --- a/.dockerignore +++ b/.dockerignore @@ -23,3 +23,9 @@ scripts/* !scripts/verify_attestation.py !pyproject.toml !vanguarstew_agent_files.json + +# Private runtime state must never enter a build context (openvang product runtime). +.env +data/ +*.sqlite3 +private-review-results/ diff --git a/.env.example b/.env.example new file mode 100644 index 00000000..45f8ce47 --- /dev/null +++ b/.env.example @@ -0,0 +1,25 @@ +# Copy to .env. Keep this file and vanguarstew.json free of real credentials. +# Existing shell environment values take priority over entries in .env. + +# Runtime safety: the initial service does no network or inference work. +VANGUARSTEW_DRY_RUN=true +VANGUARSTEW_ALLOW_EXTERNAL_INFERENCE=false +VANGUARSTEW_POLL_ENABLED=false +VANGUARSTEW_POLL_SECONDS=300 +VANGUARSTEW_MAX_JOBS_PER_CYCLE=1 +VANGUARSTEW_HOST=127.0.0.1 +VANGUARSTEW_PORT=8080 + +# GitHub is read-only. Use a least-privilege token or, in a later rollout, +# a GitHub App installation token. This runtime cannot write GitHub comments, +# labels, approvals, closures, merges, or releases. +VANGUARSTEW_GITHUB_TOKEN= +VANGUARSTEW_GITHUB_API_BASE=https://api.github.com + +# Managed inference is opt-in. Never place this value in vanguarstew.json. +VANGUARSTEW_MODEL= +VANGUARSTEW_API_BASE= +VANGUARSTEW_API_KEY= + +# Optional GitHub webhook HMAC secret. Leave blank to disable the webhook route. +VANGUARSTEW_WEBHOOK_SECRET= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 666060c3..d14ae314 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,7 +63,7 @@ jobs: - uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} - - run: python -m pip install --upgrade pip pytest pytest-cov + - run: python -m pip install --upgrade pip pytest pytest-cov cryptography - name: Test (offline) with coverage floor env: VANGUARSTEW_OFFLINE: "1" diff --git a/.gitignore b/.gitignore index 0d59c271..3a11121f 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,7 @@ venv/ .coverage .coverage.* htmlcov/ +.env +data/ +*.sqlite3 +private-review-results/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 00000000..95812e85 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,14 @@ +# Private self-hosted runtime. No secret is copied into this image. +FROM python:3.12-slim + +WORKDIR /app +COPY . /app +RUN pip install --no-cache-dir . \ + && useradd --create-home --uid 10001 vanguarstew \ + && mkdir -p /var/lib/vanguarstew \ + && chown -R vanguarstew:vanguarstew /app /var/lib/vanguarstew + +USER vanguarstew +ENV VANGUARSTEW_DATA_DIR=/var/lib/vanguarstew +EXPOSE 8080 +CMD ["vanguarstew", "serve", "--config", "/app/vanguarstew.json"] diff --git a/deploy/systemd/vanguarstew.service b/deploy/systemd/vanguarstew.service new file mode 100644 index 00000000..998f7c8e --- /dev/null +++ b/deploy/systemd/vanguarstew.service @@ -0,0 +1,23 @@ +[Unit] +Description=Vanguarstew private maintainer-assist runtime +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=vanguarstew +Group=vanguarstew +WorkingDirectory=/opt/vanguarstew +EnvironmentFile=/etc/vanguarstew/env +ExecStart=/opt/vanguarstew/.venv/bin/vanguarstew serve --config /etc/vanguarstew/vanguarstew.json +Restart=on-failure +RestartSec=10 +UMask=0077 +NoNewPrivileges=true +PrivateTmp=true +ProtectHome=true +ProtectSystem=strict +ReadWritePaths=/var/lib/vanguarstew + +[Install] +WantedBy=multi-user.target diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 00000000..887ba8d2 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,14 @@ +services: + vanguarstew: + build: . + restart: unless-stopped + env_file: + - .env + volumes: + - ./data:/var/lib/vanguarstew + - ./vanguarstew.json:/app/vanguarstew.json:ro + ports: + - "127.0.0.1:8080:8080" + init: true + security_opt: + - no-new-privileges:true diff --git a/docs/openvang-agent-factory.md b/docs/openvang-agent-factory.md new file mode 100644 index 00000000..47d469a3 --- /dev/null +++ b/docs/openvang-agent-factory.md @@ -0,0 +1,150 @@ +# OpenVang agent factory + +## Product direction + +OpenVang is building the owner-level operating system for a Bittensor subnet. +It coordinates specialist agents for validator work, maintainer stewardship, +miner QA, building and running, product planning, QA, scheduling, and +defensive adversarial QA. It is not a monolithic autonomous owner account. + +The first factory implementation is policy-only: it makes authority, memory, +and publication boundaries machine-checkable before any chain, wallet, or +repository-write adapter is introduced. + +## Roles + +| Role | Owns | May not do automatically | +| --- | --- | --- | +| Validator | validation, scoring, receipt verification | change emissions, sign, vote, publish | +| Maintainer | private repository analysis and recommendations | comment, merge, label, close, publish review evidence | +| Miner QA | miner protocol/output conformance | score itself, alter miner state, publish raw traces | +| Builder | isolated builds and bounded runs | deploy or access credentials | +| Product | plans and owner proposals | publish, change roadmap/governance state | +| QA | acceptance and regression evidence | bypass a failing gate | +| Scheduler | dispatch, leases, recovery | grant permissions or perform an owner action | +| Security QA | defensive adversarial testing and remediation proposals | exploit third parties, disclose findings, mutate production | + +“Exploit agent” is therefore implemented as **security QA**: it performs +authorized, defensive attack simulation in isolated environments and proposes +containment. It does not receive an offensive execution capability. + +## Owner boundary + +The factory has no automatic capability for: + +- wallet/key access or signing; +- Bittensor transactions, emissions changes, or governance votes; +- GitHub writes: comments, labels, reviews, closes, merges, releases, or + permission changes; and +- public communication or publication. + +An agent can prepare a commitment-bound `ActionIntent` for such work. The +intent contains digests of the payload and rationale, not an execution handle. +`FactoryPolicy.can_auto_execute(...)` is always false. A later owner-action +adapter requires its own proposal, signer separation, explicit approval, +idempotency design, audit policy, and rollback/containment plan. + +## Memory contract + +Every memory item has both a role scope and an artifact class. + +| Scope | Intended use | Cross-role transfer | +| --- | --- | --- | +| `role-private` | raw private operational material | never | +| `shared-commitment` | bounded, shaped coordination facts | commitment-only | +| `publishable-commitment` | externally verifiable public-safe fact | commitment-only | + +Private maintainer review material remains `role-private`. It cannot cross into +another role, an HTTP response, a benchmark artifact, a Polaris receipt, a +GitHub comment, or public status. A validator may create a narrow public-safe +commitment, but the policy does not give it publication authority. + +`openvang/memory.py` makes the role boundary durable for factory work. A +`FactoryMemoryVault` accepts role-private JSON only through an +operator-supplied authenticated cipher, binds ciphertext to the record id, +role, and plaintext commitment, and keeps an append-only owner-only SQLite +store. Reading requires the same role; a wrong key, altered ciphertext, or +altered binding fails closed. The default package does not invent or persist a +key: deployments using the local Fernet implementation install +`vanguarstew[private-memory]` and supply their key through a secret manager or +other operator-controlled channel. + +Cross-role memory in this vault accepts only an already-shaped SHA-256 +commitment. It has no API to derive a shared fact from a private record. This +prevents private reviewer material—including its existence, content, reasoning, +or source trace—from being promoted into another role's memory or any public +surface. + +This builds on the existing Vanguarstew memory rule: live and benchmark memory +remain separate, benchmark views are time-safe, and raw memory is excluded from +attestation evidence. + +## Initial implementation + +`openvang/factory.py` declares all eight role contracts and checks: + +1. every role has a least-privilege action set; +2. no role can gain an owner-level action; +3. only the validator can create a publishable memory commitment; +4. cross-role exchange excludes role-private artifacts; and +5. public shapes are commitment-only and no role can publish them directly. + +It deliberately has no Bittensor SDK, signer, wallet, GitHub-write client, +public webhook, or remote execution dependency. + +`openvang/scheduler.py` is the first private control-plane primitive. It stores +only opaque task/output commitments, target role, allowed action, output scope, +budget units, status, and a bounded worker lease in an owner-only SQLite file. +It cannot execute work or expose a task through a network interface. + +`openvang/isolated.py` is the first non-privileged adapter. It accepts only a +live leased `run-isolated` task for miner QA, builder, QA, or security QA. Its +input commitment, the owner-supplied approval digest, and an exact +`SealedExecutionPlan` must all match. It delegates only to the existing sealed +executor with its fixed resource and network boundary, independently checks the +aggregate-only result contract, and writes just an output digest back to the +scheduler. It has no shell-command interface, remote executor, credential +access, GitHub client, public output, or cross-role-memory access. + +The adapter is deliberately local and operator-invoked for now. A worker must +claim a lease that covers the approved execution time; a stale or hand-built +task is rejected before the sealed executor starts. Its result receipt is a +local commitment, not a Polaris receipt, publication, or proof of workload +confidentiality. + +`openvang/subnet.py` provides the next adapter boundary without embedding a +Bittensor client. It accepts an injected, separately reviewed read-only source +and a fixed `subnet-state-v1` projection: network, netuid, block height, and +participant/validator counts. The source cannot return hotkeys, wallet data, +endpoints, weights, prompts, or arbitrary fields through this adapter. A live +validator, miner-QA, or product task is bound to the exact request commitment; +only the normalized snapshot digest is retained. The operator remains +responsible for the source endpoint, credentials (if any), and independently +enforcing that it has no write capability. + +Inspect the static contract locally: + +```bash +vanguarstew factory-policy +``` + +This command is informational only. It does not load runtime configuration or +secrets, inspect memory, contact a subnet, or authorize an action. + +## Rollout sequence + +1. Run the policy registry and commitment-only scheduler beside the current + Vanguarstew runtime in dry-run mode and record only aggregate + authorization-denial telemetry locally. +2. Attach one non-privileged adapter at a time: local isolated build/QA and a + strict read-only subnet-state boundary are available; a live Bittensor + source still requires separate deployment review. A worker claims only work + assigned to its role. +3. Pilot one subnet with a fixed budget and an independent validator/QA check. +4. Only after an operator workflow and threat model are approved, consider a + narrowly scoped owner-action adapter. It must use external signing and a + human approval step; no general owner key enters the factory. + +The existing 24/7 Vanguarstew runtime remains a private maintainer-assist +component within this architecture. It is not promoted to a subnet-wide owner +agent by this policy scaffold. diff --git a/docs/product-runtime-plan.md b/docs/product-runtime-plan.md new file mode 100644 index 00000000..3f8d6c1e --- /dev/null +++ b/docs/product-runtime-plan.md @@ -0,0 +1,121 @@ +# Product runtime plan + +## Goal + +Turn Vanguarstew from an operator-run development tool into a self-hosted +maintainer-assist service that survives restarts, is straightforward to deploy, +and preserves the project's strict private-review boundary. The product keeps +the existing `solve(...)` contract and benchmark/TEE systems separate from +live operations. + +The first deliverable is a private control plane. It accepts work, stores +private results locally, and exposes only loopback health checks. It does not +post a review, merge, label, close, reopen, or otherwise mutate GitHub. + +## Security and publication contract + +| Data | Runtime handling | Public output | +| --- | --- | --- | +| GitHub token, webhook secret, model key | environment only; never JSON config, logs, or status | never | +| PR diff, model prompt, review result, private review evidence | owner-only local result directory | never | +| Runtime queue and heartbeat | owner-local SQLite | never | +| `/healthz`, `/readyz` | loopback-only operational endpoints | static health state only | +| Benchmark / Polaris TEE evidence | independent benchmark pipeline | existing receipt-safe commitments only | + +Live review data must never be copied into benchmark artifacts, Polaris receipts, +leaderboards, GitHub comments, or the runtime HTTP response. A public result +requires a separate, explicit publication design and review; it is outside this +runtime plan. + +## Delivery phases + +1. **Private local foundation — implemented now.** `vanguarstew init`, + `doctor`, `run-once`, and `serve`; env-only secrets; durable SQLite work + queue; owner-only review files; read-only GitHub client; signed webhook + intake; loopback health probes; Compose and systemd templates. +2. **Controlled live pilot.** Configure one repository and a least-privilege + GitHub App/read token. Enable inference explicitly, keep outputs local, and + observe cost, queue latency, retries, and failure classes. No automatic + GitHub write action. +3. **Operator workflow.** Add an authenticated private operator console or + explicit command for a maintainer to inspect and selectively publish a + bounded, policy-approved summary. This must not reveal private reviewer + purpose, evidence, or reasoning traces. +4. **Scale and recovery.** Move queue ownership to a managed database only if + the local SQLite deployment has demonstrated a real capacity limit; add + encrypted backup/restore drills, metrics with aggregate-only telemetry, and + key rotation. +5. **Optional automation.** Any GitHub write capability needs a separate + threat model, GitHub App permission review, idempotency contract, audit + controls, and an explicit maintainer approval gate. It is not enabled by + this implementation. + +## Operator flow + +```bash +cp .env.example .env +cp vanguarstew.json.example vanguarstew.json +python -m pip install -e . +vanguarstew doctor +vanguarstew serve +``` + +The copied configuration is intentionally inert: dry-run mode is on and polling +is off. `doctor` makes the state visible without making a network request and +without printing any secret. For a controlled pilot, the operator must make all +three conscious changes in `.env`: + +```dotenv +VANGUARSTEW_DRY_RUN=false +VANGUARSTEW_ALLOW_EXTERNAL_INFERENCE=true +VANGUARSTEW_POLL_ENABLED=true +``` + +They must also provide `VANGUARSTEW_GITHUB_TOKEN`, `VANGUARSTEW_MODEL`, +`VANGUARSTEW_API_BASE`, and `VANGUARSTEW_API_KEY`. The GitHub integration is +read-only; completing a local review still causes no GitHub mutation. + +For a service manager, use either `docker compose up -d` (the port remains +bound to `127.0.0.1`) or adapt `deploy/systemd/vanguarstew.service`. Keep the +data directory, `.env`, configuration, and journal private to the operator. + +## Operational checks + +- `vanguarstew doctor` must pass before starting the service. +- `curl http://127.0.0.1:8080/healthz` and `/readyz` are the only intended + unauthenticated monitoring probes. Neither identifies a repository, PR, or + review outcome. +- Inspect `data/private-review-results/` only on the host; it is deliberately + not an API route. +- Keep `VANGUARSTEW_DRY_RUN=true` for installation and upgrades. Explicitly + enable live inference only after validating the selected model provider's + data-handling terms and spend limit. +- The queue uses delivery/head identifiers to make repeated webhook delivery or + poll cycles harmless. Failed work remains locally visible as a failure class, + not as a published review trace. +- Work deferred by dry-run or disabled inference returns to the queue only when + the operator explicitly enables live private inference. A crashed in-progress + claim is retried only after its 15-minute lease expires; hard failures are not + retried automatically. + +## Acceptance gates for the first live pilot + +1. A fresh host can complete the operator flow from an empty data directory. +2. Restarting the process retains queued and completed state without duplicating + a delivery. +3. A valid signed webhook queues at most one review; an invalid signature + exposes no payload and creates no work. +4. A dry run produces no network requests and no inference invocation. +5. No endpoint, log line, benchmark artifact, or GitHub action exposes review + content or creates a GitHub mutation. +6. A local review result is owner-readable only, and the service still exposes + only health/readiness status. + +## Non-goals for this phase + +- Replacing the validator-facing `solve(...)` entrypoint. +- Treating Polaris as a confidentiality layer. Polaris remains an integrity + receipt path for supported benchmark jobs, not a store for live review data. +- Auto-merge, auto-close, auto-label, comment posting, or participant scoring. +- Claiming memory improves live quality before an independently held-out, + preregistered ablation passes its declared gate. diff --git a/openvang/__init__.py b/openvang/__init__.py new file mode 100644 index 00000000..d2521fb0 --- /dev/null +++ b/openvang/__init__.py @@ -0,0 +1,61 @@ +"""OpenVang's policy-only multi-agent factory foundation. + +This package deliberately defines *who may propose and prepare work*, not how +to control a wallet, submit a chain transaction, merge code, or publish a +review. Those owner-level effects remain outside the factory until a separate +approved adapter and signer boundary exists. +""" + +from .factory import ( + ActionIntent, + ActionKind, + ArtifactClass, + FactoryPolicy, + MemoryScope, + Role, + RoleContract, +) +from .isolated import IsolatedExecutionAdapter, IsolatedExecutionError, IsolatedExecutionReceipt +from .memory import ( + AuthenticatedCipher, + FactoryMemoryError, + FactoryMemoryVault, + FernetMemoryCipher, + PrivateMemoryRecord, + SharedMemoryCommitment, +) +from .scheduler import FactoryScheduler, FactoryTask, SchedulerError +from .subnet import ( + ReadOnlySubnetStateAdapter, + ReadOnlySubnetStateSource, + SubnetStateError, + SubnetStatePlan, + SubnetStateReceipt, +) + +__all__ = [ + "ActionIntent", + "ActionKind", + "ArtifactClass", + "FactoryPolicy", + "MemoryScope", + "Role", + "RoleContract", + "FactoryScheduler", + "FactoryTask", + "SchedulerError", + "IsolatedExecutionAdapter", + "IsolatedExecutionError", + "IsolatedExecutionReceipt", + "AuthenticatedCipher", + "FactoryMemoryError", + "FactoryMemoryVault", + "FernetMemoryCipher", + "PrivateMemoryRecord", + "SharedMemoryCommitment", + "ReadOnlySubnetStateAdapter", + "ReadOnlySubnetStateSource", + "SubnetStateError", + "SubnetStatePlan", + "SubnetStateReceipt", +] diff --git a/openvang/factory.py b/openvang/factory.py new file mode 100644 index 00000000..dbef25b8 --- /dev/null +++ b/openvang/factory.py @@ -0,0 +1,438 @@ +"""Explicit authority and memory policy for the OpenVang subnet-agent factory. + +The factory is a coordination layer for multiple specialist agents. It is not +a privileged owner key, a Bittensor signer, or a GitHub automation client. +Every owner-level effect is represented only as an immutable intent and is +denied automatic execution by this module. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from enum import Enum +from typing import Iterable, Mapping + + +class FactoryPolicyError(ValueError): + """A role, capability, artifact, or intent violates the factory contract.""" + + +class Role(str, Enum): + """Specialist roles operated within a single subnet's owner workflow.""" + + VALIDATOR = "validator" + MAINTAINER = "maintainer" + MINER_QA = "miner-qa" + BUILDER = "builder" + PRODUCT = "product" + QA = "qa" + SCHEDULER = "scheduler" + SECURITY_QA = "security-qa" + + +class ActionKind(str, Enum): + """Actions a role may prepare or perform in the non-privileged factory.""" + + READ_SUBNET_STATE = "read-subnet-state" + READ_REPOSITORY = "read-repository" + PLAN = "plan" + RUN_ISOLATED = "run-isolated" + VALIDATE = "validate" + SCORE = "score" + VERIFY_RECEIPT = "verify-receipt" + WRITE_PRIVATE_MEMORY = "write-private-memory" + READ_ROLE_MEMORY = "read-role-memory" + DISPATCH = "dispatch" + PROPOSE_OWNER_ACTION = "propose-owner-action" + GITHUB_WRITE = "github-write" + ONCHAIN_TRANSACTION = "onchain-transaction" + WALLET_ACCESS = "wallet-access" + EMISSION_CHANGE = "emission-change" + GOVERNANCE_VOTE = "governance-vote" + PUBLICATION = "publication" + + +class MemoryScope(str, Enum): + """Visibility classes for a role's memory projection.""" + + ROLE_PRIVATE = "role-private" + SHARED_COMMITMENT = "shared-commitment" + PUBLISHABLE_COMMITMENT = "publishable-commitment" + + +class ArtifactClass(str, Enum): + """Classes with deliberately different publication and retention rules.""" + + PRIVATE_REVIEW = "private-review" + PRIVATE_OPERATION = "private-operation" + SHARED_COMMITMENT = "shared-commitment" + PUBLIC_COMMITMENT = "public-commitment" + PUBLIC_STATUS = "public-status" + + +_AUTOMATIC_ACTIONS = frozenset( + { + ActionKind.READ_SUBNET_STATE, + ActionKind.READ_REPOSITORY, + ActionKind.PLAN, + ActionKind.RUN_ISOLATED, + ActionKind.VALIDATE, + ActionKind.SCORE, + ActionKind.VERIFY_RECEIPT, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.DISPATCH, + ActionKind.PROPOSE_OWNER_ACTION, + } +) +_OWNER_ACTIONS = frozenset( + { + ActionKind.GITHUB_WRITE, + ActionKind.ONCHAIN_TRANSACTION, + ActionKind.WALLET_ACCESS, + ActionKind.EMISSION_CHANGE, + ActionKind.GOVERNANCE_VOTE, + ActionKind.PUBLICATION, + } +) +_PUBLIC_ARTIFACTS = frozenset({ArtifactClass.PUBLIC_COMMITMENT, ArtifactClass.PUBLIC_STATUS}) + + +@dataclass(frozen=True) +class RoleContract: + """The minimum authority granted to a specialist role. + + The granted action set must contain only non-privileged capabilities. A + role can request an owner action through a commitment-only intent, but it + cannot obtain a direct effect permission from the registry. + """ + + role: Role + purpose: str + actions: frozenset[ActionKind] + readable_memory: frozenset[MemoryScope] + writable_memory: frozenset[MemoryScope] + + def __post_init__(self) -> None: + if not isinstance(self.role, Role): + raise FactoryPolicyError("role contract role must be a Role") + if not self.purpose.strip(): + raise FactoryPolicyError("role purpose must be non-empty") + if any(not isinstance(action, ActionKind) for action in self.actions): + raise FactoryPolicyError("role contract actions must be ActionKind values") + if any(not isinstance(scope, MemoryScope) for scope in self.readable_memory | self.writable_memory): + raise FactoryPolicyError("role contract memory scopes must be MemoryScope values") + if not self.actions <= _AUTOMATIC_ACTIONS: + raise FactoryPolicyError("role contracts cannot grant owner-level effects") + if not self.readable_memory or not self.writable_memory: + raise FactoryPolicyError("role contracts need explicit memory scopes") + if not self.writable_memory <= self.readable_memory: + raise FactoryPolicyError("a role may write only memory it can read") + if MemoryScope.PUBLISHABLE_COMMITMENT in self.writable_memory and self.role != Role.VALIDATOR: + raise FactoryPolicyError("only the validator may write publishable commitments") + + +@dataclass(frozen=True) +class ActionIntent: + """A non-executable, commitment-bound request for a protected owner effect.""" + + requested_by: Role + action: ActionKind + payload_commitment: str + reason_commitment: str + + def __post_init__(self) -> None: + if not isinstance(self.requested_by, Role) or not isinstance(self.action, ActionKind): + raise FactoryPolicyError("action intent role and action must use factory enums") + if self.action not in _OWNER_ACTIONS: + raise FactoryPolicyError("an action intent must request an owner-level effect") + for name, value in ( + ("payload_commitment", self.payload_commitment), + ("reason_commitment", self.reason_commitment), + ): + if not isinstance(value, str) or len(value) != 64 or any(c not in "0123456789abcdef" for c in value): + raise FactoryPolicyError(f"{name} must be a lowercase SHA-256 commitment") + + @property + def digest(self) -> str: + return _digest( + { + "requested_by": self.requested_by.value, + "action": self.action.value, + "payload_commitment": self.payload_commitment, + "reason_commitment": self.reason_commitment, + } + ) + + +def _digest(value: Mapping[str, str]) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _contract( + role: Role, + purpose: str, + actions: Iterable[ActionKind], + readable_memory: Iterable[MemoryScope], + writable_memory: Iterable[MemoryScope], +) -> RoleContract: + return RoleContract( + role=role, + purpose=purpose, + actions=frozenset(actions), + readable_memory=frozenset(readable_memory), + writable_memory=frozenset(writable_memory), + ) + + +def default_contracts() -> tuple[RoleContract, ...]: + """Return the fixed, least-privilege contracts for the initial factory.""" + private = (MemoryScope.ROLE_PRIVATE, MemoryScope.SHARED_COMMITMENT) + return ( + _contract( + Role.VALIDATOR, + "Validate subnet work, score permitted artifacts, and verify receipt-safe evidence.", + ( + ActionKind.READ_SUBNET_STATE, + ActionKind.VALIDATE, + ActionKind.SCORE, + ActionKind.VERIFY_RECEIPT, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + (*private, MemoryScope.PUBLISHABLE_COMMITMENT), + (*private, MemoryScope.PUBLISHABLE_COMMITMENT), + ), + _contract( + Role.MAINTAINER, + "Prepare repository stewardship analysis and private maintainer recommendations.", + ( + ActionKind.READ_REPOSITORY, + ActionKind.PLAN, + ActionKind.VALIDATE, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + _contract( + Role.MINER_QA, + "Exercise miner-facing behavior and verify protocol and output conformance.", + ( + ActionKind.READ_SUBNET_STATE, + ActionKind.RUN_ISOLATED, + ActionKind.VALIDATE, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + _contract( + Role.BUILDER, + "Build and run bounded workloads in an isolated execution environment.", + ( + ActionKind.READ_REPOSITORY, + ActionKind.RUN_ISOLATED, + ActionKind.VALIDATE, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + _contract( + Role.PRODUCT, + "Synthesize public product signals into plans and owner-reviewable proposals.", + ( + ActionKind.READ_SUBNET_STATE, + ActionKind.READ_REPOSITORY, + ActionKind.PLAN, + ActionKind.READ_ROLE_MEMORY, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + _contract( + Role.QA, + "Run independent acceptance, regression, and integration checks.", + ( + ActionKind.READ_REPOSITORY, + ActionKind.RUN_ISOLATED, + ActionKind.VALIDATE, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + _contract( + Role.SCHEDULER, + "Dispatch bounded work and recover queue leases without receiving owner authority.", + ( + ActionKind.DISPATCH, + ActionKind.READ_ROLE_MEMORY, + ActionKind.WRITE_PRIVATE_MEMORY, + ), + private, + private, + ), + _contract( + Role.SECURITY_QA, + "Perform defensive adversarial QA and propose containment or remediation.", + ( + ActionKind.READ_REPOSITORY, + ActionKind.RUN_ISOLATED, + ActionKind.VALIDATE, + ActionKind.WRITE_PRIVATE_MEMORY, + ActionKind.READ_ROLE_MEMORY, + ActionKind.PROPOSE_OWNER_ACTION, + ), + private, + private, + ), + ) + + +class FactoryPolicy: + """Validate actions, memory exchange, and publication before adapters exist.""" + + def __init__(self, contracts: Iterable[RoleContract] | None = None): + selected = tuple(default_contracts() if contracts is None else contracts) + if set(contract.role for contract in selected) != set(Role): + raise FactoryPolicyError("factory policy must define every role exactly once") + if len({contract.role for contract in selected}) != len(selected): + raise FactoryPolicyError("factory policy defines a role more than once") + self._contracts = {contract.role: contract for contract in selected} + + @property + def contracts(self) -> tuple[RoleContract, ...]: + return tuple(self._contracts[role] for role in Role) + + def contract(self, role: Role) -> RoleContract: + return self._contracts[Role(role)] + + def allows(self, role: Role, action: ActionKind) -> bool: + """Return whether a non-owner action is granted to this role.""" + return ActionKind(action) in self.contract(Role(role)).actions + + def assert_allowed(self, role: Role, action: ActionKind) -> None: + if not self.allows(role, action): + raise FactoryPolicyError(f"{Role(role).value} cannot perform {ActionKind(action).value}") + + def can_read_memory(self, role: Role, scope: MemoryScope) -> bool: + return MemoryScope(scope) in self.contract(Role(role)).readable_memory + + def can_write_memory(self, role: Role, scope: MemoryScope) -> bool: + return MemoryScope(scope) in self.contract(Role(role)).writable_memory + + def can_transfer_memory( + self, + *, + source: Role, + target: Role, + scope: MemoryScope, + artifact: ArtifactClass, + ) -> bool: + """Allow only commitment-level sharing between different roles. + + A role-private item—including private maintainer review evidence—never + crosses a role boundary through the factory. Cross-role collaboration + uses a shaped commitment, not raw history, prompts, or reviewer output. + """ + source = Role(source) + target = Role(target) + scope = MemoryScope(scope) + artifact = ArtifactClass(artifact) + if not self.can_read_memory(source, scope) or not self.can_read_memory(target, scope): + return False + if source == target: + return True + return scope in {MemoryScope.SHARED_COMMITMENT, MemoryScope.PUBLISHABLE_COMMITMENT} and artifact in { + ArtifactClass.SHARED_COMMITMENT, + ArtifactClass.PUBLIC_COMMITMENT, + } + + def may_publish(self, role: Role, artifact: ArtifactClass) -> bool: + """Publication is never a role capability in the initial factory.""" + Role(role) + ArtifactClass(artifact) + return False + + def public_contract(self) -> dict[str, object]: + """Return static policy metadata safe for an operator or deployment check. + + This is a registry description, not runtime telemetry: it contains no + assigned work, private memory, artifact, review, wallet, or owner data. + """ + return { + "schema_version": 1, + "roles": [ + { + "role": contract.role.value, + "actions": sorted(action.value for action in contract.actions), + "readable_memory": sorted(scope.value for scope in contract.readable_memory), + "writable_memory": sorted(scope.value for scope in contract.writable_memory), + } + for contract in self.contracts + ], + "owner_actions_require_external_approval": sorted(action.value for action in _OWNER_ACTIONS), + "automatic_owner_execution": False, + "automatic_publication": False, + } + + def public_shape_allowed(self, artifact: ArtifactClass, fields: Iterable[str]) -> bool: + """Validate the narrow class of commitment-only public artifacts. + + This rejects private-review output by construction and keeps public + evidence independent of agent prompts, memory, histories, or reasoning. + """ + artifact = ArtifactClass(artifact) + field_set = frozenset(fields) + if artifact not in _PUBLIC_ARTIFACTS: + return False + allowed = { + "schema_version", + "policy_version", + "commitment", + "status", + "verified_at", + } + return bool(field_set) and field_set <= allowed + + def intent(self, role: Role, action: ActionKind, *, payload: Mapping[str, object], reason: str) -> ActionIntent: + """Prepare an owner-reviewable request without retaining executable payload text.""" + role = Role(role) + action = ActionKind(action) + self.assert_allowed(role, ActionKind.PROPOSE_OWNER_ACTION) + if action not in _OWNER_ACTIONS: + raise FactoryPolicyError("factory intents are reserved for owner-level actions") + if not isinstance(reason, str) or not reason.strip(): + raise FactoryPolicyError("owner-action reason must be non-empty") + try: + payload_commitment = _digest({"payload": json.dumps(payload, sort_keys=True, separators=(",", ":"))}) + except (TypeError, ValueError) as exc: + raise FactoryPolicyError("owner-action payload must be JSON-compatible") from exc + return ActionIntent( + requested_by=role, + action=action, + payload_commitment=payload_commitment, + reason_commitment=hashlib.sha256(reason.strip().encode("utf-8")).hexdigest(), + ) + + def can_auto_execute(self, intent: ActionIntent) -> bool: + """Owner effects cannot be auto-executed by factory policy, ever.""" + if not isinstance(intent, ActionIntent): + raise FactoryPolicyError("execution requires an action intent") + return False diff --git a/openvang/isolated.py b/openvang/isolated.py new file mode 100644 index 00000000..e15d745f --- /dev/null +++ b/openvang/isolated.py @@ -0,0 +1,148 @@ +"""Approval-bound adapter for isolated OpenVang build and QA work. + +This module deliberately connects the factory only to the existing sealed +executor. It is not a shell runner, remote-execution client, credential +bridge, or publication mechanism. A task may run only when its factory +commitment exactly equals an owner-approved ``SealedExecutionPlan`` request. +The scheduler retains a digest of the verified aggregate envelope, never the +envelope, workload, private output, or error detail itself. +""" + +from __future__ import annotations + +import hashlib +import hmac +from dataclasses import dataclass + +from benchmark.sealed_aggregate import verify_sealed_aggregate +from benchmark.sealed_execution import SealedExecutionPlan, SealedExecutor + +from .factory import ActionKind, ArtifactClass, MemoryScope, Role +from .scheduler import FactoryScheduler, FactoryTask, SchedulerError + + +class IsolatedExecutionError(RuntimeError): + """An isolated task was not authorized, failed, or produced no safe receipt.""" + + +_ISOLATED_ROLES = frozenset({Role.MINER_QA, Role.BUILDER, Role.QA, Role.SECURITY_QA}) +_SHA256_LENGTH = 64 + + +@dataclass(frozen=True) +class IsolatedExecutionReceipt: + """Commitment-only outcome for a locally completed sealed workload.""" + + task_id: int + input_commitment: str + output_commitment: str + + +def _matches_commitment(value: object, expected: str) -> bool: + return ( + isinstance(value, str) + and len(value) == _SHA256_LENGTH + and all(character in "0123456789abcdef" for character in value) + and hmac.compare_digest(value, expected) + ) + + +class IsolatedExecutionAdapter: + """Execute one live leased build/QA task through ``SealedExecutor`` only. + + The caller must first queue a permitted ``run-isolated`` task whose input + commitment is the plan's ``request_sha256()``, then claim it using the same + role. The separate approval argument has to match that exact commitment; + neither scheduler state nor a role can self-approve a changed plan. + """ + + def __init__(self, scheduler: FactoryScheduler, *, executor: SealedExecutor | None = None): + if not isinstance(scheduler, FactoryScheduler): + raise TypeError("scheduler must be a FactoryScheduler") + if executor is not None and not isinstance(executor, SealedExecutor): + raise TypeError("executor must be a SealedExecutor") + self.scheduler = scheduler + self.executor = executor or SealedExecutor() + + def execute( + self, + task: FactoryTask, + plan: SealedExecutionPlan, + *, + approved_request_sha256: str, + ) -> IsolatedExecutionReceipt: + """Run one approved plan and store its aggregate digest as task output. + + No sealed result is returned to the caller. A pre-execution policy or + binding failure marks a verified claimed task failed without invoking + the executor. Executor and aggregate-gate failures also retain only a + fixed failure code, never a private exception or workload transcript. + """ + try: + task = self.scheduler.require_running(task) + except SchedulerError as exc: + raise IsolatedExecutionError("isolated task is not an active claimed task") from exc + + try: + self._validate_binding(task, plan, approved_request_sha256) + except IsolatedExecutionError: + self._fail(task, code="isolated-approval-rejected") + raise + + try: + envelope = self.executor.execute_approved( + plan, + approved_request_sha256=approved_request_sha256, + ) + if not isinstance(envelope, str) or not verify_sealed_aggregate( + envelope, + expected_challenge=plan.challenge, + ).get("ok"): + raise ValueError("sealed aggregate rejected") + output_commitment = hashlib.sha256(envelope.encode("utf-8")).hexdigest() + except Exception: + self._fail(task, code="sealed-execution-failed") + raise IsolatedExecutionError("sealed execution failed") from None + + try: + self.scheduler.complete( + task.id, + role=task.role, + output_commitment=output_commitment, + ) + except SchedulerError as exc: + raise IsolatedExecutionError("sealed result could not be recorded") from exc + return IsolatedExecutionReceipt( + task_id=task.id, + input_commitment=task.input_commitment, + output_commitment=output_commitment, + ) + + @staticmethod + def _validate_binding( + task: FactoryTask, + plan: SealedExecutionPlan, + approved_request_sha256: str, + ) -> None: + if task.role not in _ISOLATED_ROLES or task.action != ActionKind.RUN_ISOLATED: + raise IsolatedExecutionError("task is not authorized for isolated execution") + if ( + task.output_scope != MemoryScope.ROLE_PRIVATE + or task.output_artifact != ArtifactClass.PRIVATE_OPERATION + ): + raise IsolatedExecutionError("isolated task output is not role-private") + if not isinstance(plan, SealedExecutionPlan): + raise IsolatedExecutionError("an exact sealed execution plan is required") + request_commitment = plan.request_sha256() + if not hmac.compare_digest(task.input_commitment, request_commitment): + raise IsolatedExecutionError("sealed plan does not match task commitment") + if not _matches_commitment(approved_request_sha256, request_commitment): + raise IsolatedExecutionError("sealed plan does not have exact external approval") + + def _fail(self, task: FactoryTask, *, code: str) -> None: + try: + self.scheduler.fail(task.id, role=task.role, code=code) + except SchedulerError: + # Never replace a useful authorization/execution error with a task + # state detail. The scheduler contains no raw result either way. + pass diff --git a/openvang/memory.py b/openvang/memory.py new file mode 100644 index 00000000..eb7b09d2 --- /dev/null +++ b/openvang/memory.py @@ -0,0 +1,383 @@ +"""Encrypted role-private memory and commitment-only factory coordination. + +This store is separate from benchmark/controller memory. It is for local +factory workers and makes two boundaries durable: + +* role-private content is encrypted at rest and can be read only through the + same declared role; and +* cross-role coordination accepts only a caller-supplied SHA-256 commitment. + +There is deliberately no method that derives or exports a shared commitment +from a role-private record. In particular, private maintainer-review material +cannot create a cross-role or public trace through this vault. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import sqlite3 +import stat +import threading +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Protocol + +from .factory import ActionKind, FactoryPolicy, MemoryScope, Role + + +class FactoryMemoryError(RuntimeError): + """Factory memory storage, encryption, or boundary validation failed.""" + + +class AuthenticatedCipher(Protocol): + """AEAD-like interface supplied by the local operator or a key service.""" + + def encrypt(self, plaintext: bytes, *, associated_data: bytes) -> bytes: + """Return authenticated ciphertext bound to ``associated_data``.""" + + def decrypt(self, ciphertext: bytes, *, associated_data: bytes) -> bytes: + """Return plaintext only when ciphertext and associated data verify.""" + + +class FernetMemoryCipher: + """Optional local Fernet implementation for an operator-managed key. + + ``cryptography`` is intentionally optional so importing the base package + does not silently add a key-management dependency. Deployments that use + this adapter install ``vanguarstew[private-memory]`` and keep the Fernet + key outside the repository, scheduler, and database. + """ + + def __init__(self, key: bytes | str): + try: + from cryptography.fernet import Fernet, InvalidToken + except ImportError as exc: # pragma: no cover - depends on install extra + raise FactoryMemoryError("private memory requires vanguarstew[private-memory]") from exc + if isinstance(key, str): + key = key.encode("ascii") + if not isinstance(key, bytes): + raise FactoryMemoryError("Fernet memory key must be bytes or ASCII text") + try: + self._fernet = Fernet(key) + except (TypeError, ValueError) as exc: + raise FactoryMemoryError("Fernet memory key is invalid") from exc + self._invalid_token = InvalidToken + + @classmethod + def generate_key(cls) -> bytes: + try: + from cryptography.fernet import Fernet + except ImportError as exc: # pragma: no cover - depends on install extra + raise FactoryMemoryError("private memory requires vanguarstew[private-memory]") from exc + return Fernet.generate_key() + + def encrypt(self, plaintext: bytes, *, associated_data: bytes) -> bytes: + if not isinstance(plaintext, bytes) or not isinstance(associated_data, bytes): + raise FactoryMemoryError("private memory cipher inputs must be bytes") + # Fernet has no associated-data parameter. Prefixing a fixed-length + # domain separator and exact AAD lets decryption authenticate both as + # one token without exposing the AAD in the token plaintext to callers. + return self._fernet.encrypt(len(associated_data).to_bytes(4, "big") + associated_data + plaintext) + + def decrypt(self, ciphertext: bytes, *, associated_data: bytes) -> bytes: + if not isinstance(ciphertext, bytes) or not isinstance(associated_data, bytes): + raise FactoryMemoryError("private memory cipher inputs must be bytes") + try: + combined = self._fernet.decrypt(ciphertext) + except self._invalid_token as exc: + raise FactoryMemoryError("private memory ciphertext could not be authenticated") from exc + if len(combined) < 4: + raise FactoryMemoryError("private memory ciphertext is malformed") + length = int.from_bytes(combined[:4], "big") + bound = combined[4 : 4 + length] + plaintext = combined[4 + length :] + if len(bound) != length or not hmac.compare_digest(bound, associated_data): + raise FactoryMemoryError("private memory associated data does not match") + return plaintext + + +@dataclass(frozen=True) +class PrivateMemoryRecord: + """Metadata safe to return after writing encrypted role-private content.""" + + id: str + role: Role + commitment: str + created_at: str + + +@dataclass(frozen=True) +class SharedMemoryCommitment: + """A shaped coordination fact that contains no role-private payload.""" + + id: str + source_role: Role + commitment: str + created_at: str + + +_MAX_PRIVATE_BYTES = 64 * 1024 +_SHA256_HEX = frozenset("0123456789abcdef") + + +def _utcnow() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def _canonical_json(value: object) -> bytes: + try: + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") + except (TypeError, ValueError) as exc: + raise FactoryMemoryError("private memory content must be JSON-compatible") from exc + if len(encoded) > _MAX_PRIVATE_BYTES: + raise FactoryMemoryError("private memory content exceeds the fixed size limit") + return encoded + + +def _commitment(value: object, *, label: str) -> str: + if ( + not isinstance(value, str) + or len(value) != 64 + or any(character not in _SHA256_HEX for character in value) + ): + raise FactoryMemoryError(f"{label} must be a lowercase SHA-256 commitment") + return value + + +def _record_id(value: object) -> str: + if not isinstance(value, str) or len(value) != 32 or any(character not in _SHA256_HEX for character in value): + raise FactoryMemoryError("private memory record id is malformed") + return value + + +def _associated_data(*, record_id: str, role: Role, commitment: str) -> bytes: + return f"openvang-private-memory-v1:{record_id}:{role.value}:{commitment}".encode("ascii") + + +def _secure_directory(path: Path) -> None: + path.mkdir(mode=0o700, parents=True, exist_ok=True) + try: + path.chmod(0o700) + except OSError: + pass + + +class FactoryMemoryVault: + """Owner-local, append-only storage with explicit role and sharing policy.""" + + def __init__( + self, + database_path: str | Path, + *, + cipher: AuthenticatedCipher, + policy: FactoryPolicy | None = None, + ): + if not callable(getattr(cipher, "encrypt", None)) or not callable(getattr(cipher, "decrypt", None)): + raise TypeError("cipher must provide encrypt and decrypt") + self.database_path = Path(database_path) + self.cipher = cipher + self.policy = policy or FactoryPolicy() + _secure_directory(self.database_path.parent) + self._lock = threading.RLock() + self._connection = sqlite3.connect( + self.database_path, + timeout=30, + isolation_level=None, + check_same_thread=False, + ) + self._connection.row_factory = sqlite3.Row + self._connection.execute("PRAGMA journal_mode=DELETE") + self._initialize() + try: + self.database_path.chmod(stat.S_IRUSR | stat.S_IWUSR) + except OSError: + pass + + def _initialize(self) -> None: + with self._lock: + self._connection.executescript( + """ + CREATE TABLE IF NOT EXISTS private_memory_records ( + id TEXT PRIMARY KEY, + role TEXT NOT NULL, + commitment TEXT NOT NULL, + ciphertext BLOB NOT NULL, + created_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS private_memory_role_created + ON private_memory_records(role, created_at, id); + CREATE TABLE IF NOT EXISTS shared_memory_commitments ( + id TEXT PRIMARY KEY, + source_role TEXT NOT NULL, + commitment TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(source_role, commitment) + ); + CREATE INDEX IF NOT EXISTS shared_memory_created + ON shared_memory_commitments(created_at, id); + CREATE TRIGGER IF NOT EXISTS private_memory_records_immutable_update + BEFORE UPDATE ON private_memory_records + BEGIN SELECT RAISE(ABORT, 'private memory records are append-only'); END; + CREATE TRIGGER IF NOT EXISTS private_memory_records_immutable_delete + BEFORE DELETE ON private_memory_records + BEGIN SELECT RAISE(ABORT, 'private memory records are append-only'); END; + CREATE TRIGGER IF NOT EXISTS shared_memory_commitments_immutable_update + BEFORE UPDATE ON shared_memory_commitments + BEGIN SELECT RAISE(ABORT, 'shared memory commitments are append-only'); END; + CREATE TRIGGER IF NOT EXISTS shared_memory_commitments_immutable_delete + BEFORE DELETE ON shared_memory_commitments + BEGIN SELECT RAISE(ABORT, 'shared memory commitments are append-only'); END; + """ + ) + + def close(self) -> None: + with self._lock: + self._connection.close() + + def __enter__(self) -> "FactoryMemoryVault": + return self + + def __exit__(self, exc_type, exc, traceback) -> None: + self.close() + + def append_private(self, *, role: Role, content: object) -> PrivateMemoryRecord: + """Encrypt one bounded role-private JSON value and return its metadata.""" + role = Role(role) + self.policy.assert_allowed(role, ActionKind.WRITE_PRIVATE_MEMORY) + if not self.policy.can_write_memory(role, MemoryScope.ROLE_PRIVATE): + raise FactoryMemoryError("role cannot write role-private memory") + plaintext = _canonical_json(content) + commitment = hashlib.sha256(plaintext).hexdigest() + record_id = uuid.uuid4().hex + created_at = _utcnow() + try: + ciphertext = self.cipher.encrypt( + plaintext, + associated_data=_associated_data(record_id=record_id, role=role, commitment=commitment), + ) + except FactoryMemoryError: + raise + except Exception as exc: + raise FactoryMemoryError("private memory encryption failed") from exc + if not isinstance(ciphertext, bytes) or not ciphertext: + raise FactoryMemoryError("private memory cipher returned invalid ciphertext") + with self._lock: + self._connection.execute( + """ + INSERT INTO private_memory_records(id, role, commitment, ciphertext, created_at) + VALUES (?, ?, ?, ?, ?) + """, + (record_id, role.value, commitment, sqlite3.Binary(ciphertext), created_at), + ) + return PrivateMemoryRecord(record_id, role, commitment, created_at) + + def read_private(self, *, role: Role, record_id: str) -> object: + """Decrypt one record only for its exact role; no cross-role fallback exists.""" + role = Role(role) + self.policy.assert_allowed(role, ActionKind.READ_ROLE_MEMORY) + if not self.policy.can_read_memory(role, MemoryScope.ROLE_PRIVATE): + raise FactoryMemoryError("role cannot read role-private memory") + record_id = _record_id(record_id) + with self._lock: + row = self._connection.execute( + """ + SELECT id, role, commitment, ciphertext + FROM private_memory_records WHERE id=? AND role=? + """, + (record_id, role.value), + ).fetchone() + if row is None: + raise FactoryMemoryError("private memory record is unavailable to this role") + commitment = _commitment(row["commitment"], label="private memory commitment") + try: + plaintext = self.cipher.decrypt( + bytes(row["ciphertext"]), + associated_data=_associated_data(record_id=record_id, role=role, commitment=commitment), + ) + except FactoryMemoryError: + raise + except Exception as exc: + raise FactoryMemoryError("private memory decryption failed") from exc + if not isinstance(plaintext, bytes) or len(plaintext) > _MAX_PRIVATE_BYTES: + raise FactoryMemoryError("private memory plaintext is invalid") + if not hmac.compare_digest(hashlib.sha256(plaintext).hexdigest(), commitment): + raise FactoryMemoryError("private memory plaintext commitment does not match") + try: + value = json.loads(plaintext) + except (TypeError, ValueError) as exc: + raise FactoryMemoryError("private memory plaintext is not valid JSON") from exc + if not hmac.compare_digest(_canonical_json(value), plaintext): + raise FactoryMemoryError("private memory plaintext is not canonical JSON") + return value + + def append_shared_commitment(self, *, source_role: Role, commitment: str) -> SharedMemoryCommitment: + """Persist one already-shaped cross-role commitment without any payload.""" + source_role = Role(source_role) + self.policy.assert_allowed(source_role, ActionKind.WRITE_PRIVATE_MEMORY) + if not self.policy.can_write_memory(source_role, MemoryScope.SHARED_COMMITMENT): + raise FactoryMemoryError("role cannot write shared commitments") + commitment = _commitment(commitment, label="shared memory commitment") + record_id = uuid.uuid4().hex + created_at = _utcnow() + with self._lock: + try: + self._connection.execute( + """ + INSERT INTO shared_memory_commitments(id, source_role, commitment, created_at) + VALUES (?, ?, ?, ?) + """, + (record_id, source_role.value, commitment, created_at), + ) + except sqlite3.IntegrityError: + row = self._connection.execute( + """ + SELECT id, source_role, commitment, created_at + FROM shared_memory_commitments WHERE source_role=? AND commitment=? + """, + (source_role.value, commitment), + ).fetchone() + assert row is not None + return _shared_record(row) + return SharedMemoryCommitment(record_id, source_role, commitment, created_at) + + def shared_commitments(self, *, role: Role, limit: int = 50) -> tuple[SharedMemoryCommitment, ...]: + """Return bounded commitment-only coordination facts for an allowed role.""" + role = Role(role) + self.policy.assert_allowed(role, ActionKind.READ_ROLE_MEMORY) + if not self.policy.can_read_memory(role, MemoryScope.SHARED_COMMITMENT): + raise FactoryMemoryError("role cannot read shared commitments") + if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 100: + raise FactoryMemoryError("shared commitment limit must be between 1 and 100") + with self._lock: + rows = self._connection.execute( + """ + SELECT id, source_role, commitment, created_at + FROM shared_memory_commitments + ORDER BY created_at DESC, id DESC LIMIT ? + """, + (limit,), + ).fetchall() + return tuple(_shared_record(row) for row in rows) + + def counts(self) -> dict[str, int]: + """Return aggregate local counts only; no content, id, role, or commitment.""" + with self._lock: + private = self._connection.execute("SELECT COUNT(*) FROM private_memory_records").fetchone()[0] + shared = self._connection.execute("SELECT COUNT(*) FROM shared_memory_commitments").fetchone()[0] + return {"private_records": int(private), "shared_commitments": int(shared)} + + +def _shared_record(row: sqlite3.Row) -> SharedMemoryCommitment: + try: + return SharedMemoryCommitment( + id=_record_id(row["id"]), + source_role=Role(row["source_role"]), + commitment=_commitment(row["commitment"], label="shared memory commitment"), + created_at=str(row["created_at"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise FactoryMemoryError("shared memory record is malformed") from exc diff --git a/openvang/scheduler.py b/openvang/scheduler.py new file mode 100644 index 00000000..b5d8f6e5 --- /dev/null +++ b/openvang/scheduler.py @@ -0,0 +1,402 @@ +"""Private, role-aware task scheduler for the OpenVang factory. + +The scheduler coordinates *commitments*, not raw prompts, repository data, +credentials, review evidence, or executable owner-action payloads. It does not +run a worker itself and cannot call Bittensor, a wallet, GitHub, or a public +endpoint. A worker must claim only tasks assigned to its own factory role. +""" + +from __future__ import annotations + +import sqlite3 +import threading +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from .factory import ActionKind, ArtifactClass, FactoryPolicy, MemoryScope, Role + + +class SchedulerError(ValueError): + """The requested task lifecycle operation violates the scheduler contract.""" + + +class TaskStatus(str): + """Stored task states. Kept as strings for portable SQLite inspection.""" + + QUEUED = "queued" + RUNNING = "running" + SUCCEEDED = "succeeded" + DEFERRED = "deferred" + FAILED = "failed" + + +_ACTIVE_STATUSES = (TaskStatus.QUEUED, TaskStatus.RUNNING) +_TERMINAL_STATUSES = (TaskStatus.SUCCEEDED, TaskStatus.DEFERRED, TaskStatus.FAILED) +_VALID_STATUSES = (*_ACTIVE_STATUSES, *_TERMINAL_STATUSES) + + +def _utcnow() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def _secure_directory(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + try: + path.chmod(0o700) + except OSError: + pass + + +def _commitment(value: object, *, field: str) -> str: + if not isinstance(value, str) or len(value) != 64 or any(char not in "0123456789abcdef" for char in value): + raise SchedulerError(f"{field} must be a lowercase SHA-256 commitment") + return value + + +def _positive_integer(value: object, *, field: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise SchedulerError(f"{field} must be a positive integer") + return value + + +def _output_allowed( + policy: FactoryPolicy, + *, + role: Role, + scope: MemoryScope, + artifact: ArtifactClass, +) -> bool: + """Require output scope and artifact classification to agree exactly.""" + if not policy.can_write_memory(role, scope): + return False + if artifact == ArtifactClass.PRIVATE_REVIEW: + return role == Role.MAINTAINER and scope == MemoryScope.ROLE_PRIVATE + if artifact == ArtifactClass.PRIVATE_OPERATION: + return scope == MemoryScope.ROLE_PRIVATE + if artifact == ArtifactClass.SHARED_COMMITMENT: + return scope == MemoryScope.SHARED_COMMITMENT + if artifact == ArtifactClass.PUBLIC_COMMITMENT: + return role == Role.VALIDATOR and scope == MemoryScope.PUBLISHABLE_COMMITMENT + # Public status is a publication act, so no scheduler task may create it. + return False + + +@dataclass(frozen=True) +class FactoryTask: + """A safe task projection: all workload and result content is commitment-only.""" + + id: int + role: Role + action: ActionKind + input_commitment: str + output_scope: MemoryScope + output_artifact: ArtifactClass + budget_units: int + attempts: int + + +class FactoryScheduler: + """Durable, bounded scheduler for non-privileged factory work. + + `max_active_budget` limits the combined units of queued and running tasks. + It provides a deterministic local spending/throughput guard before an + external worker adapter exists; it is not a wallet or on-chain accounting + mechanism. + """ + + def __init__( + self, + database_path: str | Path, + *, + policy: FactoryPolicy | None = None, + max_active_budget: int = 10, + ): + self.policy = policy or FactoryPolicy() + self.database_path = Path(database_path) + self.max_active_budget = _positive_integer(max_active_budget, field="max_active_budget") + _secure_directory(self.database_path.parent) + self._lock = threading.RLock() + self._connection = sqlite3.connect( + self.database_path, + timeout=30, + isolation_level=None, + check_same_thread=False, + ) + self._connection.row_factory = sqlite3.Row + self._connection.execute("PRAGMA journal_mode=WAL") + self._initialize() + try: + self.database_path.chmod(0o600) + except OSError: + pass + + def _initialize(self) -> None: + with self._lock: + self._connection.executescript( + """ + CREATE TABLE IF NOT EXISTS factory_tasks ( + id INTEGER PRIMARY KEY, + role TEXT NOT NULL, + action TEXT NOT NULL, + input_commitment TEXT NOT NULL UNIQUE, + output_scope TEXT NOT NULL, + output_artifact TEXT NOT NULL, + budget_units INTEGER NOT NULL CHECK (budget_units > 0), + status TEXT NOT NULL CHECK (status IN + ('queued', 'running', 'succeeded', 'deferred', 'failed')), + attempts INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + lease_until TEXT, + output_commitment TEXT, + failure_code TEXT + ); + CREATE INDEX IF NOT EXISTS factory_tasks_dispatch + ON factory_tasks(role, status, created_at, id); + """ + ) + + def close(self) -> None: + with self._lock: + self._connection.close() + + def __enter__(self) -> "FactoryScheduler": + return self + + def __exit__(self, exc_type, exc, traceback) -> None: + self.close() + + def active_budget(self) -> int: + with self._lock: + row = self._connection.execute( + "SELECT COALESCE(SUM(budget_units), 0) AS total FROM factory_tasks " + "WHERE status IN ('queued', 'running')" + ).fetchone() + return int(row["total"]) + + def enqueue( + self, + *, + role: Role, + action: ActionKind, + input_commitment: str, + output_scope: MemoryScope, + output_artifact: ArtifactClass, + budget_units: int = 1, + ) -> bool: + """Queue a permitted role task once, respecting the active budget cap.""" + scheduler_role = Role.SCHEDULER + self.policy.assert_allowed(scheduler_role, ActionKind.DISPATCH) + role = Role(role) + action = ActionKind(action) + output_scope = MemoryScope(output_scope) + output_artifact = ArtifactClass(output_artifact) + input_commitment = _commitment(input_commitment, field="input_commitment") + budget_units = _positive_integer(budget_units, field="budget_units") + if not self.policy.allows(role, action): + raise SchedulerError(f"{role.value} is not permitted to perform {action.value}") + if action == ActionKind.PROPOSE_OWNER_ACTION: + raise SchedulerError("owner-action proposals are not schedulable worker tasks") + if not _output_allowed( + self.policy, + role=role, + scope=output_scope, + artifact=output_artifact, + ): + raise SchedulerError("task output scope or artifact is not permitted for this role") + now = _utcnow() + with self._lock: + self._connection.execute("BEGIN IMMEDIATE") + try: + duplicate = self._connection.execute( + "SELECT 1 FROM factory_tasks WHERE input_commitment=?", (input_commitment,) + ).fetchone() + if duplicate is not None: + self._connection.execute("COMMIT") + return False + if self.active_budget() + budget_units > self.max_active_budget: + raise SchedulerError("active budget limit would be exceeded") + self._connection.execute( + """ + INSERT INTO factory_tasks( + role, action, input_commitment, output_scope, output_artifact, + budget_units, status, attempts, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, 'queued', 0, ?, ?) + """, + ( + role.value, + action.value, + input_commitment, + output_scope.value, + output_artifact.value, + budget_units, + now, + now, + ), + ) + self._connection.execute("COMMIT") + except Exception: + self._connection.execute("ROLLBACK") + raise + return True + + def recover_expired_leases(self) -> int: + """Make work recoverable after an interrupted worker lease expires.""" + now = _utcnow() + with self._lock: + cursor = self._connection.execute( + """ + UPDATE factory_tasks + SET status='queued', lease_until=NULL, failure_code='lease-expired', updated_at=? + WHERE status='running' AND lease_until IS NOT NULL AND lease_until <= ? + """, + (now, now), + ) + return cursor.rowcount + + def claim_next(self, role: Role, *, lease_seconds: int = 300) -> FactoryTask | None: + """Claim one queued task for exactly the worker's declared role.""" + role = Role(role) + lease_seconds = _positive_integer(lease_seconds, field="lease_seconds") + self.recover_expired_leases() + now = _utcnow() + lease_until = (datetime.now(timezone.utc) + timedelta(seconds=lease_seconds)).replace( + microsecond=0 + ).isoformat() + with self._lock: + self._connection.execute("BEGIN IMMEDIATE") + try: + row = self._connection.execute( + """ + SELECT id, role, action, input_commitment, output_scope, output_artifact, + budget_units, attempts + FROM factory_tasks + WHERE role=? AND status='queued' + ORDER BY created_at, id LIMIT 1 + """, + (role.value,), + ).fetchone() + if row is None: + self._connection.execute("COMMIT") + return None + cursor = self._connection.execute( + """ + UPDATE factory_tasks + SET status='running', attempts=attempts+1, lease_until=?, updated_at=? + WHERE id=? AND status='queued' + """, + (lease_until, now, row["id"]), + ) + self._connection.execute("COMMIT") + except Exception: + self._connection.execute("ROLLBACK") + raise + if cursor.rowcount != 1: + return None + return _task_from_row(row, attempts=int(row["attempts"]) + 1) + + def require_running(self, task: FactoryTask) -> FactoryTask: + """Return the canonical task only while its matching worker lease is live. + + Adapters call this immediately before an irreversible local operation. + It prevents a hand-built or stale ``FactoryTask`` projection from being + used as authority to start work. This is a scheduler-state check, not + a credential system: worker identity and process isolation remain the + responsibility of the adapter deployment. + """ + if not isinstance(task, FactoryTask): + raise SchedulerError("a claimed FactoryTask is required") + now = _utcnow() + with self._lock: + row = self._connection.execute( + """ + SELECT id, role, action, input_commitment, output_scope, output_artifact, + budget_units, attempts + FROM factory_tasks + WHERE id=? AND role=? AND status='running' + AND lease_until IS NOT NULL AND lease_until > ? + """, + (task.id, task.role.value, now), + ).fetchone() + if row is None: + raise SchedulerError("task is not running with a live lease for this role") + current = _task_from_row(row, attempts=int(row["attempts"])) + if current != task: + raise SchedulerError("claimed task does not match the scheduler record") + return current + + def complete(self, task_id: int, *, role: Role, output_commitment: str) -> None: + """Finish a claimed task with a result digest only.""" + self._transition( + task_id, + role=role, + status=TaskStatus.SUCCEEDED, + output_commitment=_commitment(output_commitment, field="output_commitment"), + ) + + def defer(self, task_id: int, *, role: Role, code: str) -> None: + self._transition(task_id, role=role, status=TaskStatus.DEFERRED, failure_code=_code(code)) + + def fail(self, task_id: int, *, role: Role, code: str) -> None: + self._transition(task_id, role=role, status=TaskStatus.FAILED, failure_code=_code(code)) + + def _transition( + self, + task_id: int, + *, + role: Role, + status: str, + output_commitment: str | None = None, + failure_code: str | None = None, + ) -> None: + if isinstance(task_id, bool) or not isinstance(task_id, int) or task_id < 1: + raise SchedulerError("task_id must be a positive integer") + role = Role(role) + if status not in _TERMINAL_STATUSES: + raise SchedulerError("scheduler transition must be terminal") + with self._lock: + cursor = self._connection.execute( + """ + UPDATE factory_tasks + SET status=?, output_commitment=?, failure_code=?, lease_until=NULL, updated_at=? + WHERE id=? AND role=? AND status='running' + """, + (status, output_commitment, failure_code, _utcnow(), task_id, role.value), + ) + if cursor.rowcount != 1: + raise SchedulerError("task is not running for this role") + + def status_counts(self) -> dict[str, int]: + """Return aggregate local counts without task, repository, or output data.""" + with self._lock: + rows = self._connection.execute( + "SELECT status, COUNT(*) AS total FROM factory_tasks GROUP BY status" + ).fetchall() + counts = {status: 0 for status in _VALID_STATUSES} + counts.update({str(row["status"]): int(row["total"]) for row in rows}) + return counts + + +def _code(value: object) -> str: + if not isinstance(value, str) or not value or len(value) > 80: + raise SchedulerError("failure code must be a non-empty string of at most 80 characters") + if any(char.isspace() for char in value): + raise SchedulerError("failure code must not contain whitespace") + return value + + +def _task_from_row(row: sqlite3.Row, *, attempts: int) -> FactoryTask: + try: + return FactoryTask( + id=int(row["id"]), + role=Role(row["role"]), + action=ActionKind(row["action"]), + input_commitment=str(row["input_commitment"]), + output_scope=MemoryScope(row["output_scope"]), + output_artifact=ArtifactClass(row["output_artifact"]), + budget_units=int(row["budget_units"]), + attempts=attempts, + ) + except (KeyError, TypeError, ValueError) as exc: + raise SchedulerError("stored task violated the factory contract") from exc diff --git a/openvang/subnet.py b/openvang/subnet.py new file mode 100644 index 00000000..901f5960 --- /dev/null +++ b/openvang/subnet.py @@ -0,0 +1,192 @@ +"""Commitment-only adapter contract for read-only OpenVang subnet snapshots. + +The factory intentionally does not ship a Bittensor client, wallet, endpoint, +or credential. An operator may inject a separately reviewed read-only source. +This adapter binds its request to a leased scheduler task, validates a narrow +identity-free snapshot schema, and retains only its digest in scheduler state. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import re +from dataclasses import dataclass +from typing import Mapping, Protocol + +from .factory import ActionKind, ArtifactClass, MemoryScope, Role +from .scheduler import FactoryScheduler, FactoryTask, SchedulerError + + +class SubnetStateError(RuntimeError): + """A subnet state task, source, or snapshot violated the local contract.""" + + +class ReadOnlySubnetStateSource(Protocol): + """Minimal source interface for a separately deployed read-only collector.""" + + def read_snapshot(self, plan: "SubnetStatePlan") -> Mapping[str, object]: + """Return exactly one ``subnet-state-v1`` shaped snapshot.""" + + +_NETWORK_RE = re.compile(r"[a-z][a-z0-9-]{0,31}") +_READ_ONLY_ROLES = frozenset({Role.VALIDATOR, Role.MINER_QA, Role.PRODUCT}) +_SNAPSHOT_KEYS = frozenset( + { + "schema_version", + "network", + "netuid", + "block", + "participant_count", + "validator_count", + } +) + + +def _canonical_json(value: Mapping[str, object]) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def _bounded_int(value: object, *, label: str, minimum: int, maximum: int) -> int: + if isinstance(value, bool) or not isinstance(value, int) or not minimum <= value <= maximum: + raise SubnetStateError(f"{label} is outside the supported range") + return value + + +@dataclass(frozen=True) +class SubnetStatePlan: + """An exact, non-secret request for one read-only subnet state projection.""" + + network: str + netuid: int + + def __post_init__(self) -> None: + if not isinstance(self.network, str) or not _NETWORK_RE.fullmatch(self.network): + raise SubnetStateError("network must be a lowercase network identifier") + _bounded_int(self.netuid, label="netuid", minimum=0, maximum=2**32 - 1) + + def request_body(self) -> dict[str, object]: + return { + "schema_version": 1, + "projection": "subnet-state-v1", + "network": self.network, + "netuid": self.netuid, + } + + def request_sha256(self) -> str: + return hashlib.sha256(_canonical_json(self.request_body()).encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class SubnetStateReceipt: + """Commitment-only record of a completed read-only snapshot task.""" + + task_id: int + input_commitment: str + output_commitment: str + + +def _normalize_snapshot(snapshot: Mapping[str, object], *, plan: SubnetStatePlan) -> dict[str, object]: + if not isinstance(snapshot, Mapping) or set(snapshot) != _SNAPSHOT_KEYS: + raise SubnetStateError("subnet snapshot does not match the fixed projection") + if snapshot.get("schema_version") != 1: + raise SubnetStateError("subnet snapshot schema is unsupported") + network = snapshot.get("network") + if not isinstance(network, str) or not hmac.compare_digest(network, plan.network): + raise SubnetStateError("subnet snapshot network does not match its request") + netuid = _bounded_int(snapshot.get("netuid"), label="netuid", minimum=0, maximum=2**32 - 1) + if netuid != plan.netuid: + raise SubnetStateError("subnet snapshot netuid does not match its request") + block = _bounded_int(snapshot.get("block"), label="block", minimum=0, maximum=2**63 - 1) + participant_count = _bounded_int( + snapshot.get("participant_count"), + label="participant_count", + minimum=0, + maximum=1_000_000, + ) + validator_count = _bounded_int( + snapshot.get("validator_count"), + label="validator_count", + minimum=0, + maximum=participant_count, + ) + return { + "schema_version": 1, + "network": network, + "netuid": netuid, + "block": block, + "participant_count": participant_count, + "validator_count": validator_count, + } + + +class ReadOnlySubnetStateAdapter: + """Store a verified digest from one exact read-only subnet state request.""" + + def __init__(self, scheduler: FactoryScheduler, *, source: ReadOnlySubnetStateSource): + if not isinstance(scheduler, FactoryScheduler): + raise TypeError("scheduler must be a FactoryScheduler") + if not callable(getattr(source, "read_snapshot", None)): + raise TypeError("source must provide read_snapshot(plan)") + self.scheduler = scheduler + self.source = source + + def execute(self, task: FactoryTask, plan: SubnetStatePlan) -> SubnetStateReceipt: + """Collect and retain only the canonical snapshot commitment. + + The raw source response is intentionally neither returned nor written + to scheduler state. This method makes no endpoint, SDK, signer, or + credential decision; those stay in the separately deployed source. + """ + try: + task = self.scheduler.require_running(task) + except SchedulerError as exc: + raise SubnetStateError("subnet task is not an active claimed task") from exc + + try: + self._validate_binding(task, plan) + except SubnetStateError: + self._fail(task, code="subnet-read-rejected") + raise + + try: + snapshot = _normalize_snapshot(self.source.read_snapshot(plan), plan=plan) + output_commitment = hashlib.sha256(_canonical_json(snapshot).encode("utf-8")).hexdigest() + except Exception: + self._fail(task, code="subnet-read-failed") + raise SubnetStateError("read-only subnet snapshot failed") from None + + try: + self.scheduler.complete( + task.id, + role=task.role, + output_commitment=output_commitment, + ) + except SchedulerError as exc: + raise SubnetStateError("subnet snapshot could not be recorded") from exc + return SubnetStateReceipt( + task_id=task.id, + input_commitment=task.input_commitment, + output_commitment=output_commitment, + ) + + @staticmethod + def _validate_binding(task: FactoryTask, plan: SubnetStatePlan) -> None: + if task.role not in _READ_ONLY_ROLES or task.action != ActionKind.READ_SUBNET_STATE: + raise SubnetStateError("task is not authorized for read-only subnet state") + if ( + task.output_scope != MemoryScope.ROLE_PRIVATE + or task.output_artifact != ArtifactClass.PRIVATE_OPERATION + ): + raise SubnetStateError("subnet task output is not role-private") + if not isinstance(plan, SubnetStatePlan): + raise SubnetStateError("an exact subnet state plan is required") + if not hmac.compare_digest(task.input_commitment, plan.request_sha256()): + raise SubnetStateError("subnet state plan does not match task commitment") + + def _fail(self, task: FactoryTask, *, code: str) -> None: + try: + self.scheduler.fail(task.id, role=task.role, code=code) + except SchedulerError: + pass diff --git a/pyproject.toml b/pyproject.toml index 57898d21..3c13810b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,16 +23,20 @@ classifiers = [ dependencies = [] [project.optional-dependencies] -dev = ["pytest>=7", "pytest-cov>=4", "ruff>=0.4"] +dev = ["pytest>=7", "pytest-cov>=4", "ruff>=0.4", "cryptography>=41,<46"] +private-memory = ["cryptography>=41,<46"] tee = ["dcap-qvl==0.5.3"] +[project.scripts] +vanguarstew = "vanguarstew_runtime.cli:main" + [project.urls] Homepage = "https://github.com/gittensor-vanguard/vanguarstew" Repository = "https://github.com/gittensor-vanguard/vanguarstew" Issues = "https://github.com/gittensor-vanguard/vanguarstew/issues" [tool.setuptools] -packages = ["agent", "benchmark", "benchmark.judge_corpus", "benchmark.score_corpus", "scripts"] +packages = ["agent", "benchmark", "benchmark.judge_corpus", "benchmark.score_corpus", "openvang", "scripts", "vanguarstew_runtime"] [tool.pytest.ini_options] addopts = "-q" diff --git a/tests/test_openvang_factory.py b/tests/test_openvang_factory.py new file mode 100644 index 00000000..7950aad5 --- /dev/null +++ b/tests/test_openvang_factory.py @@ -0,0 +1,138 @@ +import hashlib +import json + +import pytest + +from openvang.factory import ( + ActionKind, + ArtifactClass, + FactoryPolicy, + FactoryPolicyError, + MemoryScope, + Role, + RoleContract, +) + + +def test_default_policy_declares_every_role_without_owner_effects(): + policy = FactoryPolicy() + + assert {contract.role for contract in policy.contracts} == set(Role) + for contract in policy.contracts: + assert ActionKind.GITHUB_WRITE not in contract.actions + assert ActionKind.ONCHAIN_TRANSACTION not in contract.actions + assert ActionKind.WALLET_ACCESS not in contract.actions + assert ActionKind.EMISSION_CHANGE not in contract.actions + assert ActionKind.GOVERNANCE_VOTE not in contract.actions + assert ActionKind.PUBLICATION not in contract.actions + + +def test_validator_alone_may_write_publishable_commitments(): + policy = FactoryPolicy() + + assert policy.can_write_memory(Role.VALIDATOR, MemoryScope.PUBLISHABLE_COMMITMENT) + for role in set(Role) - {Role.VALIDATOR}: + assert not policy.can_write_memory(role, MemoryScope.PUBLISHABLE_COMMITMENT) + + +def test_role_private_review_cannot_cross_role_boundary_or_be_published(): + policy = FactoryPolicy() + + assert not policy.can_transfer_memory( + source=Role.MAINTAINER, + target=Role.VALIDATOR, + scope=MemoryScope.ROLE_PRIVATE, + artifact=ArtifactClass.PRIVATE_REVIEW, + ) + assert not policy.may_publish(Role.MAINTAINER, ArtifactClass.PRIVATE_REVIEW) + assert not policy.public_shape_allowed(ArtifactClass.PRIVATE_REVIEW, {"commitment"}) + + +def test_only_commitment_level_artifact_can_cross_role_boundary(): + policy = FactoryPolicy() + + assert policy.can_transfer_memory( + source=Role.QA, + target=Role.VALIDATOR, + scope=MemoryScope.SHARED_COMMITMENT, + artifact=ArtifactClass.SHARED_COMMITMENT, + ) + assert not policy.can_transfer_memory( + source=Role.QA, + target=Role.VALIDATOR, + scope=MemoryScope.SHARED_COMMITMENT, + artifact=ArtifactClass.PRIVATE_OPERATION, + ) + assert policy.public_shape_allowed( + ArtifactClass.PUBLIC_COMMITMENT, + {"schema_version", "policy_version", "commitment", "verified_at"}, + ) + assert not policy.public_shape_allowed( + ArtifactClass.PUBLIC_COMMITMENT, + {"commitment", "review_reasoning"}, + ) + + +def test_owner_intent_is_commitment_only_and_never_auto_executable(): + policy = FactoryPolicy() + payload = {"subnet": 74, "action": "change-emissions", "amount": 0.6} + + intent = policy.intent( + Role.VALIDATOR, + ActionKind.EMISSION_CHANGE, + payload=payload, + reason="owner approval required after independent validation", + ) + + canonical_payload = json.dumps(payload, sort_keys=True, separators=(",", ":")) + expected_payload = hashlib.sha256( + json.dumps({"payload": canonical_payload}, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert intent.payload_commitment == expected_payload + assert "change-emissions" not in repr(intent) + assert policy.can_auto_execute(intent) is False + + +def test_roles_without_proposal_authority_cannot_request_owner_action(): + policy = FactoryPolicy() + + with pytest.raises(FactoryPolicyError, match="scheduler cannot perform propose-owner-action"): + policy.intent( + Role.SCHEDULER, + ActionKind.GITHUB_WRITE, + payload={"operation": "comment"}, + reason="not allowed", + ) + + +def test_public_contract_is_static_and_has_no_automatic_owner_path(): + contract = FactoryPolicy().public_contract() + + assert contract["automatic_owner_execution"] is False + assert contract["automatic_publication"] is False + assert {item["role"] for item in contract["roles"]} == {role.value for role in Role} + serialized = json.dumps(contract) + assert "private-review" not in serialized + assert "wallet-access" in contract["owner_actions_require_external_approval"] + + +def test_registry_rejects_a_role_contract_that_grants_owner_effect(): + with pytest.raises(FactoryPolicyError, match="cannot grant owner-level effects"): + RoleContract( + role=Role.VALIDATOR, + purpose="unsafe", + actions=frozenset({ActionKind.GITHUB_WRITE}), + readable_memory=frozenset({MemoryScope.ROLE_PRIVATE}), + writable_memory=frozenset({MemoryScope.ROLE_PRIVATE}), + ) + + +def test_registry_rejects_stringly_typed_contract_values(): + with pytest.raises(FactoryPolicyError, match="must be ActionKind"): + RoleContract( + role=Role.QA, + purpose="unsafe typing", + actions=frozenset({"github-write"}), + readable_memory=frozenset({MemoryScope.ROLE_PRIVATE}), + writable_memory=frozenset({MemoryScope.ROLE_PRIVATE}), + ) diff --git a/tests/test_openvang_isolated.py b/tests/test_openvang_isolated.py new file mode 100644 index 00000000..e380f356 --- /dev/null +++ b/tests/test_openvang_isolated.py @@ -0,0 +1,189 @@ +"""Tests for the commitment-only isolated build/QA adapter.""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from dataclasses import replace + +import pytest + +from benchmark.sealed_aggregate import build_sealed_aggregate +from benchmark.sealed_bundle import build_sealed_bundle +from benchmark.sealed_execution import SealedExecutionPlan, SealedExecutor +from openvang.factory import ActionKind, ArtifactClass, MemoryScope, Role +from openvang.isolated import IsolatedExecutionAdapter, IsolatedExecutionError +from openvang.scheduler import FactoryScheduler, TaskStatus + +CHALLENGE = "cd" * 32 + + +def _plan(tmp_path): + tmp_path.chmod(0o700) + source = tmp_path / "sealed-source" + source.mkdir(mode=0o700) + run = source / "run" + run.write_text("#!/bin/sh\nexit 99\n", encoding="utf-8") + run.chmod(0o700) + bundle = tmp_path / "sealed-bundle.tar" + build_sealed_bundle(source, bundle) + return SealedExecutionPlan(bundle_path=bundle, challenge=CHALLENGE) + + +def _aggregate(): + return build_sealed_aggregate( + { + "scored_repos": 2, + "skipped": 1, + "composite_mean": 0.625, + "composite_parts": {"judge_mean": 0.75, "objective_mean": 0.5}, + }, + challenge=CHALLENGE, + ) + + +def _claimed_task(tmp_path, plan): + scheduler = FactoryScheduler(tmp_path / "factory" / "scheduler.sqlite3") + scheduler.enqueue( + role=Role.QA, + action=ActionKind.RUN_ISOLATED, + input_commitment=plan.request_sha256(), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_OPERATION, + ) + task = scheduler.claim_next(Role.QA, lease_seconds=3600) + assert task is not None + return scheduler, task + + +def _adapter(scheduler, monkeypatch, result, calls): + executor = SealedExecutor() + + def execute_approved(plan, *, approved_request_sha256): + calls.append((plan, approved_request_sha256)) + if isinstance(result, Exception): + raise result + return result + + monkeypatch.setattr(executor, "execute_approved", execute_approved) + return IsolatedExecutionAdapter(scheduler, executor=executor) + + +def test_adapter_binds_claimed_task_approval_and_verified_aggregate_without_retaining_output( + tmp_path, monkeypatch +): + plan = _plan(tmp_path) + scheduler, task = _claimed_task(tmp_path, plan) + envelope = _aggregate() + calls = [] + adapter = _adapter(scheduler, monkeypatch, envelope, calls) + + receipt = adapter.execute( + task, + plan, + approved_request_sha256=plan.request_sha256(), + ) + + assert calls == [(plan, plan.request_sha256())] + assert receipt.task_id == task.id + assert receipt.input_commitment == plan.request_sha256() + assert receipt.output_commitment == hashlib.sha256(envelope.encode("utf-8")).hexdigest() + assert scheduler.status_counts()[TaskStatus.SUCCEEDED] == 1 + + row = sqlite3.connect(scheduler.database_path).execute( + "SELECT input_commitment, output_commitment, failure_code FROM factory_tasks WHERE id=?", (task.id,) + ).fetchone() + assert row == (plan.request_sha256(), receipt.output_commitment, None) + assert envelope not in str(row) + scheduler.close() + + +def test_adapter_rejects_changed_or_unapproved_plan_before_execution(tmp_path, monkeypatch): + plan = _plan(tmp_path) + scheduler, task = _claimed_task(tmp_path, plan) + calls = [] + adapter = _adapter(scheduler, monkeypatch, _aggregate(), calls) + + with pytest.raises(IsolatedExecutionError, match="exact external approval"): + adapter.execute(task, plan, approved_request_sha256="00" * 32) + + assert calls == [] + assert scheduler.status_counts()[TaskStatus.FAILED] == 1 + row = sqlite3.connect(scheduler.database_path).execute( + "SELECT failure_code FROM factory_tasks WHERE id=?", (task.id,) + ).fetchone() + assert row == ("isolated-approval-rejected",) + scheduler.close() + + +def test_adapter_rejects_a_plan_with_a_different_request_commitment(tmp_path, monkeypatch): + plan = _plan(tmp_path) + changed_plan = SealedExecutionPlan( + bundle_path=plan.bundle_path, + challenge=plan.challenge, + timeout_seconds=plan.timeout_seconds + 1, + ) + scheduler, task = _claimed_task(tmp_path, plan) + calls = [] + adapter = _adapter(scheduler, monkeypatch, _aggregate(), calls) + + with pytest.raises(IsolatedExecutionError, match="does not match task commitment"): + adapter.execute( + task, + changed_plan, + approved_request_sha256=changed_plan.request_sha256(), + ) + + assert calls == [] + assert scheduler.status_counts()[TaskStatus.FAILED] == 1 + scheduler.close() + + +def test_adapter_refuses_forged_or_stale_task_before_execution(tmp_path, monkeypatch): + plan = _plan(tmp_path) + scheduler, task = _claimed_task(tmp_path, plan) + calls = [] + adapter = _adapter(scheduler, monkeypatch, _aggregate(), calls) + forged = replace(task, input_commitment="ef" * 32) + + with pytest.raises(IsolatedExecutionError, match="active claimed"): + adapter.execute(forged, plan, approved_request_sha256=plan.request_sha256()) + + assert calls == [] + assert scheduler.status_counts()[TaskStatus.RUNNING] == 1 + scheduler.close() + + +def test_adapter_refuses_an_expired_lease_before_execution(tmp_path, monkeypatch): + plan = _plan(tmp_path) + scheduler, task = _claimed_task(tmp_path, plan) + scheduler._connection.execute( + "UPDATE factory_tasks SET lease_until=? WHERE id=?", ("2000-01-01T00:00:00+00:00", task.id) + ) + calls = [] + adapter = _adapter(scheduler, monkeypatch, _aggregate(), calls) + + with pytest.raises(IsolatedExecutionError, match="active claimed"): + adapter.execute(task, plan, approved_request_sha256=plan.request_sha256()) + + assert calls == [] + assert scheduler.status_counts()[TaskStatus.RUNNING] == 1 + scheduler.close() + + +def test_adapter_never_persists_invalid_or_failed_sealed_output(tmp_path, monkeypatch): + plan = _plan(tmp_path) + scheduler, task = _claimed_task(tmp_path, plan) + calls = [] + adapter = _adapter(scheduler, monkeypatch, '{"private":"not-an-aggregate"}', calls) + + with pytest.raises(IsolatedExecutionError, match="sealed execution failed"): + adapter.execute(task, plan, approved_request_sha256=plan.request_sha256()) + + assert calls == [(plan, plan.request_sha256())] + assert scheduler.status_counts()[TaskStatus.FAILED] == 1 + row = sqlite3.connect(scheduler.database_path).execute( + "SELECT output_commitment, failure_code FROM factory_tasks WHERE id=?", (task.id,) + ).fetchone() + assert row == (None, "sealed-execution-failed") + scheduler.close() diff --git a/tests/test_openvang_memory.py b/tests/test_openvang_memory.py new file mode 100644 index 00000000..fb27d8b8 --- /dev/null +++ b/tests/test_openvang_memory.py @@ -0,0 +1,80 @@ +"""Contract tests for encrypted role-private OpenVang factory memory.""" + +from __future__ import annotations + +import hashlib +import sqlite3 +import stat + +import pytest + +from openvang.factory import Role +from openvang.memory import FactoryMemoryError, FactoryMemoryVault, FernetMemoryCipher + + +def _vault(tmp_path, *, key=None): + cipher = FernetMemoryCipher(key or FernetMemoryCipher.generate_key()) + return FactoryMemoryVault(tmp_path / "factory-memory" / "vault.sqlite3", cipher=cipher), cipher + + +def test_role_private_memory_is_encrypted_append_only_and_role_scoped(tmp_path): + vault, _cipher = _vault(tmp_path) + private_content = {"review": "private-review-marker", "decision": "request changes"} + record = vault.append_private(role=Role.MAINTAINER, content=private_content) + + assert record.role == Role.MAINTAINER + assert record.commitment == hashlib.sha256( + b'{"decision":"request changes","review":"private-review-marker"}' + ).hexdigest() + assert vault.read_private(role=Role.MAINTAINER, record_id=record.id) == private_content + with pytest.raises(FactoryMemoryError, match="unavailable to this role"): + vault.read_private(role=Role.QA, record_id=record.id) + with pytest.raises(sqlite3.DatabaseError, match="append-only"): + vault._connection.execute("UPDATE private_memory_records SET role='qa' WHERE id=?", (record.id,)) + + assert b"private-review-marker" not in vault.database_path.read_bytes() + assert stat.S_IMODE(vault.database_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(vault.database_path.parent.stat().st_mode) == 0o700 + assert vault.counts() == {"private_records": 1, "shared_commitments": 0} + vault.close() + + +def test_private_memory_rejects_a_wrong_key_without_exposing_content(tmp_path): + vault, _cipher = _vault(tmp_path) + record = vault.append_private(role=Role.SECURITY_QA, content={"finding": "private-marker"}) + vault.close() + + wrong_key = FernetMemoryCipher.generate_key() + reopened = FactoryMemoryVault(vault.database_path, cipher=FernetMemoryCipher(wrong_key)) + with pytest.raises(FactoryMemoryError, match="could not be authenticated"): + reopened.read_private(role=Role.SECURITY_QA, record_id=record.id) + reopened.close() + + +def test_cross_role_coordination_accepts_only_shaped_commitments(tmp_path): + vault, _cipher = _vault(tmp_path) + private_record = vault.append_private(role=Role.MAINTAINER, content={"note": "do not share"}) + commitment = "ab" * 32 + + shared = vault.append_shared_commitment(source_role=Role.MAINTAINER, commitment=commitment) + duplicate = vault.append_shared_commitment(source_role=Role.MAINTAINER, commitment=commitment) + + assert duplicate == shared + assert vault.shared_commitments(role=Role.QA) == (shared,) + with pytest.raises(FactoryMemoryError, match="SHA-256 commitment"): + vault.append_shared_commitment(source_role=Role.MAINTAINER, commitment=private_record.id) + with pytest.raises(sqlite3.DatabaseError, match="append-only"): + vault._connection.execute( + "UPDATE shared_memory_commitments SET source_role='qa' WHERE id=?", (shared.id,) + ) + assert b"do not share" not in vault.database_path.read_bytes() + vault.close() + + +def test_private_memory_rejects_noncanonical_or_oversized_content(tmp_path): + vault, _cipher = _vault(tmp_path) + with pytest.raises(FactoryMemoryError, match="JSON-compatible"): + vault.append_private(role=Role.BUILDER, content={"unsupported": {1, 2}}) + with pytest.raises(FactoryMemoryError, match="size limit"): + vault.append_private(role=Role.BUILDER, content={"large": "x" * (64 * 1024)}) + vault.close() diff --git a/tests/test_openvang_scheduler.py b/tests/test_openvang_scheduler.py new file mode 100644 index 00000000..93da8139 --- /dev/null +++ b/tests/test_openvang_scheduler.py @@ -0,0 +1,146 @@ +import sqlite3 +import stat + +import pytest + +from openvang.factory import ActionKind, ArtifactClass, MemoryScope, Role +from openvang.scheduler import FactoryScheduler, SchedulerError, TaskStatus + + +def _digest(letter): + return letter * 64 + + +def _scheduler(tmp_path, *, budget=4): + return FactoryScheduler(tmp_path / "factory" / "scheduler.sqlite3", max_active_budget=budget) + + +def test_scheduler_dispatches_only_to_the_target_role_and_keeps_commitments(tmp_path): + with _scheduler(tmp_path) as scheduler: + assert scheduler.enqueue( + role=Role.MAINTAINER, + action=ActionKind.VALIDATE, + input_commitment=_digest("a"), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_REVIEW, + budget_units=2, + ) + assert scheduler.claim_next(Role.VALIDATOR) is None + + task = scheduler.claim_next(Role.MAINTAINER) + assert task is not None + assert task.input_commitment == _digest("a") + assert task.output_artifact == ArtifactClass.PRIVATE_REVIEW + with pytest.raises(SchedulerError, match="not running for this role"): + scheduler.complete(task.id, role=Role.QA, output_commitment=_digest("b")) + scheduler.complete(task.id, role=Role.MAINTAINER, output_commitment=_digest("b")) + + assert scheduler.status_counts()[TaskStatus.SUCCEEDED] == 1 + columns = { + row[1] + for row in sqlite3.connect(scheduler.database_path).execute("PRAGMA table_info(factory_tasks)") + } + assert "payload" not in columns + assert "output" not in columns + + +def test_scheduler_rejects_owner_effects_and_invalid_output_boundaries(tmp_path): + with _scheduler(tmp_path) as scheduler: + with pytest.raises(SchedulerError, match="not permitted"): + scheduler.enqueue( + role=Role.VALIDATOR, + action=ActionKind.ONCHAIN_TRANSACTION, + input_commitment=_digest("a"), + output_scope=MemoryScope.PUBLISHABLE_COMMITMENT, + output_artifact=ArtifactClass.PUBLIC_COMMITMENT, + ) + with pytest.raises(SchedulerError, match="output scope or artifact"): + scheduler.enqueue( + role=Role.MAINTAINER, + action=ActionKind.VALIDATE, + input_commitment=_digest("b"), + output_scope=MemoryScope.SHARED_COMMITMENT, + output_artifact=ArtifactClass.PRIVATE_REVIEW, + ) + with pytest.raises(SchedulerError, match="not schedulable"): + scheduler.enqueue( + role=Role.MAINTAINER, + action=ActionKind.PROPOSE_OWNER_ACTION, + input_commitment=_digest("c"), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_REVIEW, + ) + + +def test_scheduler_enforces_active_budget_and_deduplicates_input_commitments(tmp_path): + with _scheduler(tmp_path, budget=2) as scheduler: + kwargs = { + "role": Role.BUILDER, + "action": ActionKind.RUN_ISOLATED, + "input_commitment": _digest("a"), + "output_scope": MemoryScope.ROLE_PRIVATE, + "output_artifact": ArtifactClass.PRIVATE_OPERATION, + "budget_units": 2, + } + assert scheduler.enqueue(**kwargs) + assert not scheduler.enqueue(**kwargs) + with pytest.raises(SchedulerError, match="active budget"): + scheduler.enqueue( + role=Role.QA, + action=ActionKind.RUN_ISOLATED, + input_commitment=_digest("b"), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_OPERATION, + ) + + task = scheduler.claim_next(Role.BUILDER) + scheduler.complete(task.id, role=Role.BUILDER, output_commitment=_digest("c")) + assert scheduler.active_budget() == 0 + assert scheduler.enqueue( + role=Role.QA, + action=ActionKind.RUN_ISOLATED, + input_commitment=_digest("b"), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_OPERATION, + ) + + +def test_scheduler_recovers_expired_lease_and_keeps_storage_owner_only(tmp_path): + with _scheduler(tmp_path) as scheduler: + assert scheduler.enqueue( + role=Role.SECURITY_QA, + action=ActionKind.RUN_ISOLATED, + input_commitment=_digest("a"), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_OPERATION, + ) + first = scheduler.claim_next(Role.SECURITY_QA, lease_seconds=300) + scheduler._connection.execute( + "UPDATE factory_tasks SET lease_until=? WHERE id=?", ("2000-01-01T00:00:00+00:00", first.id) + ) + + second = scheduler.claim_next(Role.SECURITY_QA) + assert second is not None + assert second.id == first.id + assert second.attempts == 2 + assert stat.S_IMODE(scheduler.database_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(scheduler.database_path.parent.stat().st_mode) == 0o700 + + +def test_only_validator_can_schedule_a_staged_public_commitment(tmp_path): + with _scheduler(tmp_path) as scheduler: + assert scheduler.enqueue( + role=Role.VALIDATOR, + action=ActionKind.VERIFY_RECEIPT, + input_commitment=_digest("a"), + output_scope=MemoryScope.PUBLISHABLE_COMMITMENT, + output_artifact=ArtifactClass.PUBLIC_COMMITMENT, + ) + with pytest.raises(SchedulerError, match="output scope or artifact"): + scheduler.enqueue( + role=Role.QA, + action=ActionKind.VALIDATE, + input_commitment=_digest("b"), + output_scope=MemoryScope.PUBLISHABLE_COMMITMENT, + output_artifact=ArtifactClass.PUBLIC_COMMITMENT, + ) diff --git a/tests/test_openvang_subnet.py b/tests/test_openvang_subnet.py new file mode 100644 index 00000000..802d4201 --- /dev/null +++ b/tests/test_openvang_subnet.py @@ -0,0 +1,110 @@ +"""Tests for the commitment-only read-only subnet adapter.""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 + +import pytest + +from openvang.factory import ActionKind, ArtifactClass, MemoryScope, Role +from openvang.scheduler import FactoryScheduler, TaskStatus +from openvang.subnet import ReadOnlySubnetStateAdapter, SubnetStateError, SubnetStatePlan + + +class _Source: + def __init__(self, snapshot): + self.snapshot = snapshot + self.calls = [] + + def read_snapshot(self, plan): + self.calls.append(plan) + if isinstance(self.snapshot, Exception): + raise self.snapshot + return self.snapshot + + +def _plan(*, network="finney", netuid=42): + return SubnetStatePlan(network=network, netuid=netuid) + + +def _snapshot(plan): + return { + "schema_version": 1, + "network": plan.network, + "netuid": plan.netuid, + "block": 123_456, + "participant_count": 17, + "validator_count": 5, + } + + +def _claimed_task(tmp_path, plan, *, role=Role.VALIDATOR): + scheduler = FactoryScheduler(tmp_path / "factory" / "scheduler.sqlite3") + scheduler.enqueue( + role=role, + action=ActionKind.READ_SUBNET_STATE, + input_commitment=plan.request_sha256(), + output_scope=MemoryScope.ROLE_PRIVATE, + output_artifact=ArtifactClass.PRIVATE_OPERATION, + ) + task = scheduler.claim_next(role, lease_seconds=300) + assert task is not None + return scheduler, task + + +def test_read_only_adapter_binds_snapshot_to_role_task_and_keeps_only_digest(tmp_path): + plan = _plan() + snapshot = _snapshot(plan) + scheduler, task = _claimed_task(tmp_path, plan) + source = _Source(snapshot) + adapter = ReadOnlySubnetStateAdapter(scheduler, source=source) + + receipt = adapter.execute(task, plan) + + canonical = json.dumps(snapshot, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + assert source.calls == [plan] + assert receipt.output_commitment == hashlib.sha256(canonical.encode("utf-8")).hexdigest() + assert scheduler.status_counts()[TaskStatus.SUCCEEDED] == 1 + row = sqlite3.connect(scheduler.database_path).execute( + "SELECT input_commitment, output_commitment, failure_code FROM factory_tasks WHERE id=?", (task.id,) + ).fetchone() + assert row == (plan.request_sha256(), receipt.output_commitment, None) + assert canonical not in str(row) + scheduler.close() + + +def test_read_only_adapter_rejects_changed_request_before_calling_source(tmp_path): + plan = _plan() + changed_plan = _plan(netuid=43) + scheduler, task = _claimed_task(tmp_path, plan, role=Role.MINER_QA) + source = _Source(_snapshot(changed_plan)) + adapter = ReadOnlySubnetStateAdapter(scheduler, source=source) + + with pytest.raises(SubnetStateError, match="does not match task commitment"): + adapter.execute(task, changed_plan) + + assert source.calls == [] + assert scheduler.status_counts()[TaskStatus.FAILED] == 1 + scheduler.close() + + +def test_read_only_adapter_rejects_unprojected_identity_data_without_persisting_it(tmp_path): + plan = _plan() + scheduler, task = _claimed_task(tmp_path, plan, role=Role.PRODUCT) + snapshot = _snapshot(plan) + snapshot["hotkey"] = "forbidden-identity-marker" + source = _Source(snapshot) + adapter = ReadOnlySubnetStateAdapter(scheduler, source=source) + + with pytest.raises(SubnetStateError, match="snapshot failed"): + adapter.execute(task, plan) + + assert source.calls == [plan] + row = sqlite3.connect(scheduler.database_path).execute( + "SELECT output_commitment, failure_code FROM factory_tasks WHERE id=?", (task.id,) + ).fetchone() + assert row == (None, "subnet-read-failed") + assert "forbidden-identity-marker" not in str(row) + scheduler.close() diff --git a/tests/test_runtime_cli.py b/tests/test_runtime_cli.py new file mode 100644 index 00000000..ba769311 --- /dev/null +++ b/tests/test_runtime_cli.py @@ -0,0 +1,26 @@ +import json + +from vanguarstew_runtime.cli import main + + +def test_init_and_doctor_are_local_and_secret_free(tmp_path, capsys): + config_path = tmp_path / "vanguarstew.json" + + assert main(["init", "--config", str(config_path)]) == 0 + assert config_path.exists() + assert main(["doctor", "--config", str(config_path), "--env-file", str(tmp_path / "missing.env")]) == 0 + + output = capsys.readouterr().out.splitlines() + result = json.loads(output[-1]) + assert result["ok"] is True + assert result["checks"]["mode"] == "dry-run" + assert "api_key" not in json.dumps(result).lower() + + +def test_factory_policy_command_is_static_and_has_no_owner_execution(capsys): + assert main(["factory-policy"]) == 0 + + result = json.loads(capsys.readouterr().out) + assert result["automatic_owner_execution"] is False + assert result["automatic_publication"] is False + assert len(result["roles"]) == 8 diff --git a/tests/test_runtime_config.py b/tests/test_runtime_config.py new file mode 100644 index 00000000..a08fb702 --- /dev/null +++ b/tests/test_runtime_config.py @@ -0,0 +1,74 @@ +import json + +import pytest + +from vanguarstew_runtime.config import ConfigError, load_dotenv, load_runtime_config + + +def _config(path): + path.write_text( + json.dumps( + { + "version": 1, + "runtime": { + "data_dir": "runtime-data", + "host": "127.0.0.1", + "port": 8080, + "poll_seconds": 60, + "max_jobs_per_cycle": 2, + "poll_enabled": True, + }, + "repositories": [{"name": "owner/repository", "enabled": True}], + } + ) + ) + + +def test_load_runtime_config_keeps_secrets_out_of_json(tmp_path): + config_path = tmp_path / "vanguarstew.json" + _config(config_path) + + config = load_runtime_config( + config_path, + environ={ + "VANGUARSTEW_DRY_RUN": "false", + "VANGUARSTEW_ALLOW_EXTERNAL_INFERENCE": "true", + "VANGUARSTEW_GITHUB_TOKEN": "token", + "VANGUARSTEW_MODEL": "model", + "VANGUARSTEW_API_BASE": "https://example.test/v1", + "VANGUARSTEW_API_KEY": "key", + }, + ) + + assert config.data_dir == tmp_path / "runtime-data" + assert config.poll_enabled is True + assert config.dry_run is False + assert config.can_run_inference is True + assert config.enabled_repositories[0].name == "owner/repository" + + +def test_load_dotenv_does_not_evaluate_or_override_existing_environment(tmp_path): + env_file = tmp_path / ".env" + env_file.write_text("VANGUARSTEW_API_KEY='from-file'\nVALUE=$(not-executed)\n") + environment = {"VANGUARSTEW_API_KEY": "from-process"} + + load_dotenv(env_file, environment) + + assert environment["VANGUARSTEW_API_KEY"] == "from-process" + assert environment["VALUE"] == "$(not-executed)" + + +def test_runtime_config_rejects_non_https_github_endpoint(tmp_path): + config_path = tmp_path / "vanguarstew.json" + _config(config_path) + + with pytest.raises(ConfigError, match="must use https"): + load_runtime_config(config_path, environ={"VANGUARSTEW_GITHUB_API_BASE": "http://bad"}) + + +def test_runtime_config_rejects_public_http_bind(tmp_path): + config_path = tmp_path / "vanguarstew.json" + _config(config_path) + + with pytest.raises(ConfigError, match="loopback-only"): + load_runtime_config(config_path, environ={"VANGUARSTEW_HOST": "0.0.0.0"}) diff --git a/tests/test_runtime_github.py b/tests/test_runtime_github.py new file mode 100644 index 00000000..a7066f4a --- /dev/null +++ b/tests/test_runtime_github.py @@ -0,0 +1,35 @@ +import pytest + +from vanguarstew_runtime.github import GitHubClient, GitHubError + + +class _PagingClient(GitHubClient): + def __init__(self, pages): + super().__init__("https://api.example.test") + self.pages = pages + self.paths = [] + + def _get_json(self, path): + self.paths.append(path) + page = int(path.rsplit("=", 1)[1]) + return self.pages[page - 1] + + +def test_paginated_read_includes_all_pages(): + first_page = [{"number": index} for index in range(100)] + client = _PagingClient([first_page, [{"number": 100}]]) + + rows = client.list_open_pull_requests("owner/repository") + + assert len(rows) == 101 + assert client.paths == [ + "/repos/owner/repository/pulls?state=open&per_page=100&page=1", + "/repos/owner/repository/pulls?state=open&per_page=100&page=2", + ] + + +def test_paginated_read_fails_closed_at_safe_limit(): + client = _PagingClient([[{"number": 1}] * 100 for _ in range(30)]) + + with pytest.raises(GitHubError, match="safe page limit"): + client.list_open_pull_requests("owner/repository") diff --git a/tests/test_runtime_packaging.py b/tests/test_runtime_packaging.py new file mode 100644 index 00000000..24191382 --- /dev/null +++ b/tests/test_runtime_packaging.py @@ -0,0 +1,11 @@ +from pathlib import Path + + +def test_private_runtime_files_are_excluded_from_git_and_docker_contexts(): + root = Path(__file__).resolve().parents[1] + gitignore = (root / ".gitignore").read_text() + dockerignore = (root / ".dockerignore").read_text() + + for entry in (".env", "data/", "*.sqlite3", "private-review-results/"): + assert entry in gitignore + assert entry in dockerignore diff --git a/tests/test_runtime_service.py b/tests/test_runtime_service.py new file mode 100644 index 00000000..6f876f0f --- /dev/null +++ b/tests/test_runtime_service.py @@ -0,0 +1,158 @@ +import hashlib +import hmac +import json +from dataclasses import replace +from http.client import HTTPConnection + +import pytest + +from vanguarstew_runtime.config import load_runtime_config +from vanguarstew_runtime.service import RuntimeService, make_http_server +from vanguarstew_runtime.state import RuntimeState + + +class _NoNetworkGitHub: + def list_open_pull_requests(self, repository): + raise AssertionError("dry run must not poll GitHub") + + def fetch_pull_request(self, repository, number): + raise AssertionError("dry run must not fetch GitHub") + + +class _LiveGitHub: + def list_open_pull_requests(self, repository): + return [] + + def fetch_pull_request(self, repository, number): + return { + "number": number, + "title": "Local fixture", + "body": "", + "author": "contributor", + "additions": 1, + "deletions": 0, + "files": ["agent/example.py"], + "diff": "diff --git a/a b/a", + "head_sha": "head-1", + } + + +class _Reviewer: + def review(self, pull_request): + return {"action": "comment", "summary": "stored only locally"} + + +def _config(tmp_path, *, dry_run=True, webhook_secret=None): + path = tmp_path / "vanguarstew.json" + path.write_text( + json.dumps( + { + "version": 1, + "runtime": {"data_dir": "data", "poll_enabled": True, "poll_seconds": 1}, + "repositories": [{"name": "owner/repository", "enabled": True}], + } + ) + ) + environment = { + "VANGUARSTEW_DRY_RUN": str(dry_run).lower(), + "VANGUARSTEW_ALLOW_EXTERNAL_INFERENCE": "true", + "VANGUARSTEW_MODEL": "test-model", + "VANGUARSTEW_API_BASE": "https://example.test/v1", + "VANGUARSTEW_API_KEY": "test-key", + } + if webhook_secret: + environment["VANGUARSTEW_WEBHOOK_SECRET"] = webhook_secret + return load_runtime_config(path, environ=environment) + + +def test_dry_run_never_calls_github_or_inference(tmp_path): + config = _config(tmp_path, dry_run=True) + with RuntimeState(config.database_path, config.private_result_dir) as state: + state.enqueue_pull_request(delivery_id="event", repository="owner/repository", pr_number=4) + service = RuntimeService(config, state, github=_NoNetworkGitHub()) + + assert service.run_once() == {"queued": 0, "processed": 1} + assert state.queue_counts()["deferred"] == 1 + assert not list(config.private_result_dir.iterdir()) + + +def test_live_private_review_writes_no_public_result(tmp_path): + config = _config(tmp_path, dry_run=False) + with RuntimeState(config.database_path, config.private_result_dir) as state: + state.enqueue_pull_request(delivery_id="event", repository="owner/repository", pr_number=5) + service = RuntimeService(config, state, github=_LiveGitHub(), reviewer=_Reviewer()) + + assert service.run_once() == {"queued": 0, "processed": 1} + assert state.queue_counts()["succeeded"] == 1 + results = list(config.private_result_dir.glob("*.json")) + assert len(results) == 1 + assert "stored only locally" in results[0].read_text() + + +def test_explicit_live_enablement_requeues_a_dry_run_job(tmp_path): + dry_config = _config(tmp_path, dry_run=True) + live_config = _config(tmp_path, dry_run=False) + with RuntimeState(dry_config.database_path, dry_config.private_result_dir) as state: + state.enqueue_pull_request(delivery_id="event", repository="owner/repository", pr_number=5) + RuntimeService(dry_config, state, github=_NoNetworkGitHub()).run_once() + assert state.queue_counts()["deferred"] == 1 + + RuntimeService(live_config, state, github=_LiveGitHub(), reviewer=_Reviewer()).run_once() + assert state.queue_counts()["succeeded"] == 1 + + +def test_signed_webhook_is_deduplicated_and_health_exposes_no_queue(tmp_path): + secret = "webhook-secret" + config = replace(_config(tmp_path, dry_run=True, webhook_secret=secret), port=0) + with RuntimeState(config.database_path, config.private_result_dir) as state: + service = RuntimeService(config, state, github=_NoNetworkGitHub()) + body = json.dumps( + { + "action": "opened", + "number": 6, + "repository": {"full_name": "owner/repository"}, + "pull_request": {"head": {"sha": "head-6"}}, + } + ).encode() + signature = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + assert service.receive_webhook(body=body, signature=signature, event="pull_request") + assert not service.receive_webhook(body=body, signature=signature, event="pull_request") + + try: + server = make_http_server(service) + except PermissionError: + pytest.skip("test environment forbids loopback sockets") + thread = __import__("threading").Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + port = server.server_address[1] + connection = HTTPConnection("127.0.0.1", port, timeout=2) + connection.request("GET", "/healthz") + response = connection.getresponse() + payload = response.read().decode() + assert response.status == 200 + assert "repository" not in payload + assert "queued" not in payload + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +def test_signed_webhook_ignores_repositories_outside_local_allow_list(tmp_path): + secret = "webhook-secret" + config = _config(tmp_path, dry_run=True, webhook_secret=secret) + with RuntimeState(config.database_path, config.private_result_dir) as state: + service = RuntimeService(config, state, github=_NoNetworkGitHub()) + body = json.dumps( + { + "action": "opened", + "number": 7, + "repository": {"full_name": "other/repository"}, + "pull_request": {"head": {"sha": "head-7"}}, + } + ).encode() + signature = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() + + assert not service.receive_webhook(body=body, signature=signature, event="pull_request") + assert state.queue_counts()["queued"] == 0 diff --git a/tests/test_runtime_state.py b/tests/test_runtime_state.py new file mode 100644 index 00000000..23897d85 --- /dev/null +++ b/tests/test_runtime_state.py @@ -0,0 +1,69 @@ +import json +import stat + +import pytest + +from vanguarstew_runtime.state import RuntimeState + + +def _state(tmp_path): + return RuntimeState(tmp_path / "data" / "runtime.sqlite3", tmp_path / "data" / "private") + + +def test_state_deduplicates_delivery_and_claims_once(tmp_path): + with _state(tmp_path) as state: + assert state.enqueue_pull_request( + delivery_id="delivery-1", repository="owner/repo", pr_number=7, head_sha="abc" + ) + assert not state.enqueue_pull_request( + delivery_id="delivery-1", repository="owner/repo", pr_number=7, head_sha="abc" + ) + + job = state.claim_next() + assert job is not None + assert job.pr_number == 7 + assert state.claim_next() is None + state.defer(job.id, code="dry-run") + assert state.queue_counts() == { + "queued": 0, + "running": 0, + "succeeded": 0, + "deferred": 1, + "failed": 0, + } + + +def test_state_private_result_is_owner_readable_and_not_in_queue_counts(tmp_path): + with _state(tmp_path) as state: + assert state.enqueue_pull_request(delivery_id="delivery-2", repository="owner/repo", pr_number=8) + job = state.claim_next() + path_name = state.write_private_result(job.id, {"summary": "private review"}) + state.complete(job.id, result_path=path_name) + + result_path = state.private_result_dir / path_name + assert json.loads(result_path.read_text()) == {"summary": "private review"} + assert stat.S_IMODE(result_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(state.private_result_dir.stat().st_mode) == 0o700 + assert state.queue_counts()["succeeded"] == 1 + + +def test_state_refuses_terminal_transition_for_non_running_job(tmp_path): + with _state(tmp_path) as state: + with pytest.raises(ValueError, match="not running"): + state.fail(999, code="missing") + + +def test_state_recovers_expired_running_claim_but_not_a_fresh_one(tmp_path): + with _state(tmp_path) as state: + state.enqueue_pull_request(delivery_id="stale", repository="owner/repo", pr_number=9) + state.enqueue_pull_request(delivery_id="fresh", repository="owner/repo", pr_number=10) + stale = state.claim_next() + fresh = state.claim_next() + state._connection.execute( + "UPDATE jobs SET claimed_at=? WHERE id=?", ("2000-01-01T00:00:00+00:00", stale.id) + ) + + assert state.recover_expired_claims(lease_seconds=3600) == 1 + assert state.queue_counts()["queued"] == 1 + assert state.queue_counts()["running"] == 1 + assert fresh.id != stale.id diff --git a/vanguarstew.json.example b/vanguarstew.json.example new file mode 100644 index 00000000..0a70761e --- /dev/null +++ b/vanguarstew.json.example @@ -0,0 +1,17 @@ +{ + "version": 1, + "runtime": { + "data_dir": "./data", + "host": "127.0.0.1", + "port": 8080, + "poll_seconds": 300, + "max_jobs_per_cycle": 1, + "poll_enabled": false + }, + "repositories": [ + { + "name": "openvang/vanguarstew", + "enabled": true + } + ] +} diff --git a/vanguarstew_runtime/__init__.py b/vanguarstew_runtime/__init__.py new file mode 100644 index 00000000..2ae64257 --- /dev/null +++ b/vanguarstew_runtime/__init__.py @@ -0,0 +1,15 @@ +"""Private, self-hosted runtime for Vanguarstew maintainer assistance. + +The runtime is deliberately separate from the benchmark package. It persists +operational state locally, receives or polls for pull-request work, and never +publishes reviewer output. It does not add a second agent entrypoint: review +execution still uses :mod:`agent.review` and the project's managed-inference +contract. +""" + +from .config import RuntimeConfig, load_runtime_config +from .service import RuntimeService +from .state import RuntimeState + +__all__ = ["RuntimeConfig", "RuntimeService", "RuntimeState", "load_runtime_config"] + diff --git a/vanguarstew_runtime/cli.py b/vanguarstew_runtime/cli.py new file mode 100644 index 00000000..973e0ea8 --- /dev/null +++ b/vanguarstew_runtime/cli.py @@ -0,0 +1,156 @@ +"""Operator CLI for the private Vanguarstew runtime.""" + +from __future__ import annotations + +import argparse +import json +import logging +import sys +from pathlib import Path + +from .config import ( + DEFAULT_CONFIG_NAME, + DEFAULT_ENV_NAME, + ConfigError, + default_config, + load_dotenv, + load_runtime_config, +) +from .service import RuntimeService, serve_with_http +from .state import RuntimeState + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="vanguarstew", + description="self-hosted, private maintainer-assist runtime", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + init = subparsers.add_parser("init", help="write a safe starter configuration") + init.add_argument("--config", default=DEFAULT_CONFIG_NAME) + init.add_argument("--force", action="store_true", help="replace an existing configuration") + subparsers.add_parser( + "factory-policy", + help="render the static OpenVang role and authority contract", + ) + for command, help_text in ( + ("doctor", "validate local configuration without a network request"), + ("run-once", "run one bounded private work cycle"), + ("serve", "run the private worker and loopback health server"), + ): + command_parser = subparsers.add_parser(command, help=help_text) + command_parser.add_argument("--config", default=DEFAULT_CONFIG_NAME) + command_parser.add_argument("--env-file", default=DEFAULT_ENV_NAME) + serve = subparsers.choices["serve"] + serve.add_argument("--once", action="store_true", help="run one cycle without the HTTP server") + return parser + + +def _write_config(path: Path, *, force: bool) -> None: + if path.exists() and not force: + raise ConfigError(f"refusing to overwrite existing configuration: {path}") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(default_config(), indent=2) + "\n", encoding="utf-8") + try: + path.chmod(0o600) + except OSError: + pass + + +def _doctor(config_path: str, env_file: str) -> int: + environment = dict() + try: + # Preserve current process values over dotenv values, while allowing + # tests and embedding callers to pass an empty environment predictably. + import os + + environment.update(os.environ) + load_dotenv(Path(env_file), environment) + config = load_runtime_config(config_path, environ=environment) + except ConfigError as exc: + print(json.dumps({"ok": False, "error": str(exc)}, separators=(",", ":"))) + return 1 + checks = { + "configuration": "ok", + "data_directory": "ok" if config.data_dir.exists() or config.data_dir.parent.exists() else "will-create", + "github_read_token": "configured" if config.github_token else "not-configured", + "inference": "enabled" if config.can_run_inference else "not-enabled", + "mode": "dry-run" if config.dry_run else "live-private", + "polling": "enabled" if config.poll_enabled else "disabled", + "webhook": "configured" if config.webhook_secret else "disabled", + } + print(json.dumps({"ok": True, "checks": checks}, separators=(",", ":"))) + return 0 + + +def _load(config_path: str, env_file: str): + import os + + environment = dict(os.environ) + load_dotenv(Path(env_file), environment) + config = load_runtime_config(config_path, environ=environment) + state = RuntimeState(config.database_path, config.private_result_dir) + return config, state + + +def _run_once(config_path: str, env_file: str) -> int: + try: + config, state = _load(config_path, env_file) + except ConfigError as exc: + print(str(exc), file=sys.stderr) + return 1 + try: + result = RuntimeService(config, state).run_once() + print(json.dumps(result, separators=(",", ":"))) + return 0 + finally: + state.close() + + +def _serve(config_path: str, env_file: str, *, once: bool) -> int: + try: + config, state = _load(config_path, env_file) + except ConfigError as exc: + print(str(exc), file=sys.stderr) + return 1 + service = RuntimeService(config, state) + try: + if once: + print(json.dumps(service.run_once(), separators=(",", ":"))) + return 0 + serve_with_http(service) + return 0 + except KeyboardInterrupt: + return 0 + finally: + state.close() + + +def main(argv: list[str] | None = None) -> int: + """Run the operator CLI and return a process-compatible exit code.""" + args = _parser().parse_args(argv) + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + if args.command == "init": + try: + _write_config(Path(args.config), force=args.force) + except ConfigError as exc: + print(str(exc), file=sys.stderr) + return 1 + print(f"created {Path(args.config)}") + return 0 + if args.command == "factory-policy": + from openvang.factory import FactoryPolicy + + print(json.dumps(FactoryPolicy().public_contract(), sort_keys=True, separators=(",", ":"))) + return 0 + if args.command == "doctor": + return _doctor(args.config, args.env_file) + if args.command == "run-once": + return _run_once(args.config, args.env_file) + if args.command == "serve": + return _serve(args.config, args.env_file, once=args.once) + raise AssertionError("unreachable command") + + +if __name__ == "__main__": # pragma: no cover - console entry point + raise SystemExit(main()) diff --git a/vanguarstew_runtime/config.py b/vanguarstew_runtime/config.py new file mode 100644 index 00000000..ca9e7667 --- /dev/null +++ b/vanguarstew_runtime/config.py @@ -0,0 +1,299 @@ +"""Configuration for the self-hosted Vanguarstew runtime. + +Secrets are intentionally environment-only. The JSON configuration records +only non-secret operational policy, so it can be inspected and versioned +without leaking an inference credential, GitHub token, webhook secret, or +private-review content. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Mapping + +DEFAULT_CONFIG_NAME = "vanguarstew.json" +DEFAULT_ENV_NAME = ".env" + + +class ConfigError(ValueError): + """Raised when an operator configuration is missing or unsafe.""" + + +def _as_bool(value: object, *, field: str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str) and value.strip().lower() in {"1", "true", "yes", "on"}: + return True + if isinstance(value, str) and value.strip().lower() in {"0", "false", "no", "off"}: + return False + raise ConfigError(f"{field} must be a boolean") + + +def _as_positive_int(value: object, *, field: str, minimum: int = 1) -> int: + if isinstance(value, bool): + raise ConfigError(f"{field} must be an integer") + try: + number = int(value) + except (TypeError, ValueError) as exc: + raise ConfigError(f"{field} must be an integer") from exc + if number < minimum: + raise ConfigError(f"{field} must be at least {minimum}") + return number + + +def _as_string(value: object, *, field: str, default: str | None = None) -> str: + if value is None and default is not None: + return default + if not isinstance(value, str) or not value.strip(): + raise ConfigError(f"{field} must be a non-empty string") + return value.strip() + + +def _repository_name(value: object) -> str: + name = _as_string(value, field="repositories[].name") + parts = name.split("/") + if len(parts) != 2 or not all(parts): + raise ConfigError("repositories[].name must be in owner/repository form") + if any(part in {".", ".."} or " " in part for part in parts): + raise ConfigError("repositories[].name must be a GitHub owner/repository") + return name + + +def load_dotenv(path: Path, environ: dict[str, str] | None = None) -> dict[str, str]: + """Load a small, predictable dotenv file without evaluating shell syntax. + + Existing environment values always win. Shell interpolation, command + substitution, and ``export`` directives are intentionally unsupported: + configuration should never execute while it is being read. + """ + target = environ if environ is not None else os.environ + if not path.exists(): + return target + if not path.is_file(): + raise ConfigError(f"environment file is not a file: {path}") + try: + lines = path.read_text(encoding="utf-8").splitlines() + except OSError as exc: + raise ConfigError(f"cannot read environment file: {path}") from exc + for line_number, raw in enumerate(lines, start=1): + line = raw.strip() + if not line or line.startswith("#"): + continue + if line.startswith("export ") or "=" not in line: + raise ConfigError(f"invalid dotenv entry on line {line_number}") + key, value = line.split("=", 1) + key = key.strip() + value = value.strip() + if not key or not key.replace("_", "a").isalnum() or not key[0].isalpha() and key[0] != "_": + raise ConfigError(f"invalid dotenv variable on line {line_number}") + if (value.startswith('"') and value.endswith('"')) or ( + value.startswith("'") and value.endswith("'") + ): + value = value[1:-1] + target.setdefault(key, value) + return target + + +@dataclass(frozen=True) +class RepositoryTarget: + """A repository which the local runtime may read from GitHub.""" + + name: str + enabled: bool = True + + +@dataclass(frozen=True) +class RuntimeConfig: + """Validated non-secret runtime policy plus environment-backed credentials.""" + + config_path: Path + data_dir: Path + host: str + port: int + poll_seconds: int + max_jobs_per_cycle: int + poll_enabled: bool + dry_run: bool + allow_external_inference: bool + repositories: tuple[RepositoryTarget, ...] + github_api_base: str + github_token: str | None + webhook_secret: str | None + model: str | None + api_base: str | None + api_key: str | None + + @property + def database_path(self) -> Path: + return self.data_dir / "runtime.sqlite3" + + @property + def private_result_dir(self) -> Path: + return self.data_dir / "private-review-results" + + @property + def can_run_inference(self) -> bool: + return bool( + self.allow_external_inference + and self.model + and self.api_base + and self.api_key + and self.api_key != "offline" + ) + + @property + def enabled_repositories(self) -> tuple[RepositoryTarget, ...]: + return tuple(repo for repo in self.repositories if repo.enabled) + + +def _env_string(environ: Mapping[str, str], name: str, default: str | None = None) -> str | None: + value = environ.get(name, default) + if value is None: + return None + value = value.strip() + return value or None + + +def _env_bool(environ: Mapping[str, str], name: str, default: bool) -> bool: + value = environ.get(name) + return default if value is None else _as_bool(value, field=name) + + +def _env_int(environ: Mapping[str, str], name: str, default: int, *, minimum: int = 1) -> int: + value = environ.get(name) + return default if value is None else _as_positive_int(value, field=name, minimum=minimum) + + +def load_runtime_config( + config_path: str | Path = DEFAULT_CONFIG_NAME, + *, + environ: Mapping[str, str] | None = None, +) -> RuntimeConfig: + """Read and validate a non-secret JSON configuration and environment values.""" + source_env: Mapping[str, str] = os.environ if environ is None else environ + path = Path(config_path).expanduser().resolve() + if not path.exists(): + raise ConfigError( + f"configuration file not found: {path}; run `vanguarstew init --config {path}`" + ) + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ConfigError(f"configuration is not valid JSON: {path}") from exc + except OSError as exc: + raise ConfigError(f"cannot read configuration file: {path}") from exc + if not isinstance(raw, dict): + raise ConfigError("configuration root must be a JSON object") + if raw.get("version") != 1: + raise ConfigError("configuration version must be 1") + + runtime = raw.get("runtime", {}) + if not isinstance(runtime, dict): + raise ConfigError("runtime must be an object") + configured_data_dir = _as_string(runtime.get("data_dir", "./data"), field="runtime.data_dir") + env_data_dir = _env_string(source_env, "VANGUARSTEW_DATA_DIR") + data_dir = Path(env_data_dir or configured_data_dir).expanduser() + if not data_dir.is_absolute(): + data_dir = (path.parent / data_dir).resolve() + + raw_repositories = raw.get("repositories", []) + if not isinstance(raw_repositories, list): + raise ConfigError("repositories must be an array") + repositories = [] + seen_repositories = set() + for entry in raw_repositories: + if not isinstance(entry, dict): + raise ConfigError("each repository entry must be an object") + name = _repository_name(entry.get("name")) + if name.lower() in seen_repositories: + raise ConfigError(f"repository appears more than once: {name}") + seen_repositories.add(name.lower()) + repositories.append( + RepositoryTarget( + name=name, + enabled=_as_bool(entry.get("enabled", True), field=f"repositories[{name}].enabled"), + ) + ) + + host = _env_string(source_env, "VANGUARSTEW_HOST") or _as_string( + runtime.get("host", "127.0.0.1"), field="runtime.host" + ) + if host != "127.0.0.1": + raise ConfigError("runtime.host must be 127.0.0.1; private review endpoints are loopback-only") + port = _env_int( + source_env, + "VANGUARSTEW_PORT", + _as_positive_int(runtime.get("port", 8080), field="runtime.port", minimum=1), + minimum=1, + ) + if port > 65535: + raise ConfigError("runtime.port must be at most 65535") + poll_seconds = _env_int( + source_env, + "VANGUARSTEW_POLL_SECONDS", + _as_positive_int(runtime.get("poll_seconds", 300), field="runtime.poll_seconds"), + ) + max_jobs_per_cycle = _env_int( + source_env, + "VANGUARSTEW_MAX_JOBS_PER_CYCLE", + _as_positive_int( + runtime.get("max_jobs_per_cycle", 1), field="runtime.max_jobs_per_cycle" + ), + ) + poll_enabled = _env_bool( + source_env, + "VANGUARSTEW_POLL_ENABLED", + _as_bool(runtime.get("poll_enabled", False), field="runtime.poll_enabled"), + ) + dry_run = _env_bool(source_env, "VANGUARSTEW_DRY_RUN", True) + allow_external_inference = _env_bool(source_env, "VANGUARSTEW_ALLOW_EXTERNAL_INFERENCE", False) + + github_api_base = ( + _env_string(source_env, "VANGUARSTEW_GITHUB_API_BASE", "https://api.github.com") + or "https://api.github.com" + ).rstrip("/") + if not github_api_base.startswith("https://"): + raise ConfigError("VANGUARSTEW_GITHUB_API_BASE must use https") + + return RuntimeConfig( + config_path=path, + data_dir=data_dir, + host=host, + port=port, + poll_seconds=poll_seconds, + max_jobs_per_cycle=max_jobs_per_cycle, + poll_enabled=poll_enabled, + dry_run=dry_run, + allow_external_inference=allow_external_inference, + repositories=tuple(repositories), + github_api_base=github_api_base, + github_token=_env_string(source_env, "VANGUARSTEW_GITHUB_TOKEN"), + webhook_secret=_env_string(source_env, "VANGUARSTEW_WEBHOOK_SECRET"), + model=_env_string(source_env, "VANGUARSTEW_MODEL"), + api_base=_env_string(source_env, "VANGUARSTEW_API_BASE"), + api_key=_env_string(source_env, "VANGUARSTEW_API_KEY"), + ) + + +def default_config() -> dict: + """Return the safe bootstrap configuration written by ``vanguarstew init``.""" + return { + "version": 1, + "runtime": { + "data_dir": "./data", + "host": "127.0.0.1", + "port": 8080, + "poll_seconds": 300, + "max_jobs_per_cycle": 1, + "poll_enabled": False, + }, + "repositories": [ + { + "name": "openvang/vanguarstew", + "enabled": True, + } + ], + } diff --git a/vanguarstew_runtime/github.py b/vanguarstew_runtime/github.py new file mode 100644 index 00000000..23836e1e --- /dev/null +++ b/vanguarstew_runtime/github.py @@ -0,0 +1,112 @@ +"""Minimal read-only GitHub client used by the private runtime. + +There are intentionally no methods for comments, labels, closing, approving, +merging, releases, or other mutations. The runtime can gather a PR for a +local review but cannot publish or act on that review. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import quote +from urllib.request import Request, urlopen + + +class GitHubError(RuntimeError): + """A read-only GitHub request could not be completed.""" + + +@dataclass(frozen=True) +class GitHubClient: + api_base: str + token: str | None = None + timeout: float = 30.0 + + def _get_json(self, path: str) -> Any: + payload = self._get(path, accept="application/vnd.github+json") + try: + return json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise GitHubError("GitHub returned malformed JSON") from exc + + def _get(self, path: str, *, accept: str) -> bytes: + headers = { + "Accept": accept, + "X-GitHub-Api-Version": "2022-11-28", + "User-Agent": "vanguarstew-runtime/0.8", + } + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + request = Request(f"{self.api_base.rstrip('/')}{path}", headers=headers, method="GET") + try: + with urlopen(request, timeout=self.timeout) as response: + return response.read() + except HTTPError as exc: + if exc.code in {401, 403}: + raise GitHubError("GitHub authentication or rate limit rejected the read request") from exc + if exc.code == 404: + raise GitHubError("GitHub pull request or repository was not found") from exc + raise GitHubError(f"GitHub read request failed with HTTP {exc.code}") from exc + except (URLError, OSError) as exc: + raise GitHubError("GitHub read request could not reach the API") from exc + + def _get_paginated_array(self, path: str, *, maximum_pages: int = 30) -> list[dict[str, Any]]: + """Read a complete bounded GitHub list rather than silently reviewing a prefix. + + A pull request with more than one files page must not be reviewed from + only its first hundred paths. The bound avoids turning one event into + an unbounded sequence of API requests; reaching it fails closed. + """ + separator = "&" if "?" in path else "?" + result = [] + for page in range(1, maximum_pages + 1): + data = self._get_json(f"{path}{separator}page={page}") + if not isinstance(data, list): + raise GitHubError("GitHub list response was not an array") + result.extend(entry for entry in data if isinstance(entry, dict)) + if len(data) < 100: + return result + raise GitHubError("GitHub list exceeds the configured safe page limit") + + @staticmethod + def _path_repository(repository: str) -> str: + owner, name = repository.split("/", 1) + return f"/repos/{quote(owner, safe='')}/{quote(name, safe='')}" + + def list_open_pull_requests(self, repository: str) -> list[dict[str, Any]]: + return self._get_paginated_array( + f"{self._path_repository(repository)}/pulls?state=open&per_page=100" + ) + + def fetch_pull_request(self, repository: str, number: int) -> dict[str, Any]: + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + raise ValueError("pull-request number must be a positive integer") + base = f"{self._path_repository(repository)}/pulls/{number}" + data = self._get_json(base) + if not isinstance(data, dict): + raise GitHubError("GitHub pull-request response was not an object") + files_data = self._get_paginated_array(f"{base}/files?per_page=100") + diff = self._get(base, accept="application/vnd.github.v3.diff") + author = data.get("user") + login = author.get("login") if isinstance(author, dict) else None + files = [ + entry.get("filename") + for entry in files_data + if isinstance(entry, dict) and isinstance(entry.get("filename"), str) + ] + head = data.get("head") + head_sha = head.get("sha") if isinstance(head, dict) and isinstance(head.get("sha"), str) else None + return { + "number": data.get("number", number), + "title": data.get("title", ""), + "body": data.get("body"), + "author": login or "ghost", + "additions": data.get("additions", 0), + "deletions": data.get("deletions", 0), + "files": files, + "diff": diff.decode("utf-8", errors="replace"), + "head_sha": head_sha, + } diff --git a/vanguarstew_runtime/service.py b/vanguarstew_runtime/service.py new file mode 100644 index 00000000..9669399f --- /dev/null +++ b/vanguarstew_runtime/service.py @@ -0,0 +1,295 @@ +"""Private, durable execution loop for self-hosted maintainer assistance.""" + +from __future__ import annotations + +import hmac +import json +import logging +import threading +from hashlib import sha256 +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Protocol + +from agent.llm import LLM +from agent.review import review_pr + +from .config import RuntimeConfig +from .github import GitHubClient, GitHubError +from .state import ReviewJob, RuntimeState + +logger = logging.getLogger(__name__) + + +class PullRequestReader(Protocol): + """Read-only source of pull-request metadata and diffs.""" + + def list_open_pull_requests(self, repository: str) -> list[dict[str, Any]]: + ... + + def fetch_pull_request(self, repository: str, number: int) -> dict[str, Any]: + ... + + +class ReviewExecutor(Protocol): + """A private review implementation that returns a structured local result.""" + + def review(self, pull_request: dict[str, Any]) -> dict[str, Any]: + ... + + +class AgentReviewExecutor: + """Run the existing maintainer-assist review against managed inference.""" + + def __init__(self, config: RuntimeConfig): + self._config = config + + def review(self, pull_request: dict[str, Any]) -> dict[str, Any]: + llm = LLM( + model=self._config.model, + api_base=self._config.api_base, + api_key=self._config.api_key, + ) + return review_pr(pull_request, None, llm) + + +class RuntimeService: + """Poll and process review work while retaining all reviewer output locally. + + Safety is enforced in the execution flow, rather than only documented: + + * a dry run never calls GitHub or an inference provider; + * live inference requires an explicit environment opt-in; and + * no execution path sends a review result back to GitHub or a public API. + """ + + def __init__( + self, + config: RuntimeConfig, + state: RuntimeState, + *, + github: PullRequestReader | None = None, + reviewer: ReviewExecutor | None = None, + ): + self.config = config + self.state = state + self.github = github or GitHubClient(config.github_api_base, config.github_token) + self.reviewer = reviewer or AgentReviewExecutor(config) + self._stop_event = threading.Event() + + @property + def is_ready(self) -> bool: + return self.state.database_path.exists() + + def request_stop(self) -> None: + self._stop_event.set() + + def receive_webhook(self, *, body: bytes, signature: str | None, event: str | None) -> bool: + """Validate and enqueue an actionable GitHub pull-request delivery. + + False means the delivery was valid but irrelevant or already seen. No + review information is returned to the caller. + """ + secret = self.config.webhook_secret + if not secret: + raise PermissionError("webhook receiver is disabled") + expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, sha256).hexdigest() + if not signature or not hmac.compare_digest(expected, signature): + raise PermissionError("webhook signature did not match") + if event != "pull_request": + return False + try: + payload = json.loads(body.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError("webhook body was not valid JSON") from exc + if not isinstance(payload, dict) or payload.get("action") not in { + "opened", + "reopened", + "synchronize", + "ready_for_review", + }: + return False + repository = payload.get("repository") + pull_request = payload.get("pull_request") + if not isinstance(repository, dict) or not isinstance(pull_request, dict): + raise ValueError("webhook did not contain pull-request metadata") + full_name = repository.get("full_name") + number = payload.get("number") + allowed_repositories = {target.name.lower() for target in self.config.enabled_repositories} + if not isinstance(full_name, str) or full_name.lower() not in allowed_repositories: + # A correctly signed delivery can still be for another installation + # or repository. Do not turn it into private review work unless it + # is explicitly in the local allow-list. + return False + head = pull_request.get("head") + head_sha = head.get("sha") if isinstance(head, dict) else None + delivery_id = sha256(body).hexdigest() + return self.state.enqueue_pull_request( + delivery_id=f"webhook:{delivery_id}", + repository=full_name, + pr_number=number, + head_sha=head_sha if isinstance(head_sha, str) else None, + ) + + def poll_once(self) -> int: + """Queue current open PR heads using read-only GitHub API requests.""" + if not self.config.poll_enabled or self.config.dry_run: + return 0 + queued = 0 + for target in self.config.enabled_repositories: + try: + pull_requests = self.github.list_open_pull_requests(target.name) + except GitHubError: + # Keep this deliberately repository-free: runtime logs must not + # become a public trace of private reviewer activity. + logger.warning("GitHub polling failed; the runtime will retry next cycle") + continue + for pull_request in pull_requests: + number = pull_request.get("number") + head = pull_request.get("head") + head_sha = head.get("sha") if isinstance(head, dict) else None + if isinstance(number, bool) or not isinstance(number, int) or number <= 0: + continue + delivery_id = f"poll:{target.name}:{number}:{head_sha or 'unknown'}" + if self.state.enqueue_pull_request( + delivery_id=delivery_id, + repository=target.name, + pr_number=number, + head_sha=head_sha if isinstance(head_sha, str) else None, + ): + queued += 1 + return queued + + def process_one(self) -> str | None: + """Process at most one job and return only its terminal status.""" + job = self.state.claim_next() + if job is None: + return None + if self.config.dry_run: + self.state.defer(job.id, code="dry-run") + return "deferred" + if not self.config.can_run_inference: + self.state.defer(job.id, code="inference-not-explicitly-enabled") + return "deferred" + return self._review(job) + + def _review(self, job: ReviewJob) -> str: + try: + pull_request = self.github.fetch_pull_request(job.repository, job.pr_number) + review = self.reviewer.review(pull_request) + result = { + "schema_version": 1, + "review": review, + "reviewed_head_sha": pull_request.get("head_sha"), + } + result_path = self.state.write_private_result(job.id, result) + self.state.complete(job.id, result_path=result_path) + return "succeeded" + except GitHubError: + logger.warning("GitHub read failed while preparing local review") + self.state.fail(job.id, code="github-read-failed") + return "failed" + except (OSError, ValueError, TypeError): + # This covers transport/model failures and malformed provider output + # without emitting a PR identifier or review content to logs. + logger.warning("Private review did not complete") + self.state.fail(job.id, code="private-review-failed") + return "failed" + + def run_once(self) -> dict[str, int]: + """Record a heartbeat and complete a bounded amount of work.""" + self.state.recover_expired_claims() + if not self.config.dry_run and self.config.can_run_inference: + self.state.requeue_deferred() + self.state.heartbeat() + queued = self.poll_once() + processed = 0 + for _ in range(self.config.max_jobs_per_cycle): + if self.process_one() is None: + break + processed += 1 + return {"queued": queued, "processed": processed} + + def serve_forever(self) -> None: + """Run bounded cycles until an operator or signal requests shutdown.""" + while not self._stop_event.is_set(): + self.run_once() + self._stop_event.wait(self.config.poll_seconds) + + +def make_http_server(service: RuntimeService) -> ThreadingHTTPServer: + """Create a loopback-oriented health and webhook server. + + There is purposely no endpoint for queue entries, review decisions, source + evidence, prompts, or result files. Health endpoints contain static state + only and can safely be used by a local process supervisor. + """ + + class RuntimeHandler(BaseHTTPRequestHandler): + server_version = "VanguarstewRuntime/0.8" + + def log_message(self, format: str, *args: object) -> None: + # The default request log includes URLs and remote addresses. Do + # not create an operational activity trail by default. + return + + def _json(self, status: HTTPStatus, payload: dict[str, object]) -> None: + encoded = json.dumps(payload, separators=(",", ":")).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Cache-Control", "no-store") + self.send_header("Content-Length", str(len(encoded))) + self.end_headers() + self.wfile.write(encoded) + + def do_GET(self) -> None: # noqa: N802 - HTTP handler API + if self.path == "/healthz": + self._json(HTTPStatus.OK, {"ok": True, "service": "vanguarstew"}) + return + if self.path == "/readyz": + self._json(HTTPStatus.OK if service.is_ready else HTTPStatus.SERVICE_UNAVAILABLE, {"ok": service.is_ready}) + return + self._json(HTTPStatus.NOT_FOUND, {"ok": False}) + + def do_POST(self) -> None: # noqa: N802 - HTTP handler API + if self.path != "/webhooks/github": + self._json(HTTPStatus.NOT_FOUND, {"ok": False}) + return + raw_length = self.headers.get("Content-Length") + try: + length = int(raw_length or "") + except ValueError: + self._json(HTTPStatus.BAD_REQUEST, {"ok": False}) + return + if length < 1 or length > 1_000_000: + self._json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"ok": False}) + return + body = self.rfile.read(length) + try: + accepted = service.receive_webhook( + body=body, + signature=self.headers.get("X-Hub-Signature-256"), + event=self.headers.get("X-GitHub-Event"), + ) + except PermissionError: + self._json(HTTPStatus.UNAUTHORIZED, {"ok": False}) + return + except ValueError: + self._json(HTTPStatus.BAD_REQUEST, {"ok": False}) + return + self._json(HTTPStatus.ACCEPTED, {"ok": accepted}) + + return ThreadingHTTPServer((service.config.host, service.config.port), RuntimeHandler) + + +def serve_with_http(service: RuntimeService) -> None: + """Serve local health/webhook requests while the private worker runs.""" + server = make_http_server(service) + thread = threading.Thread(target=server.serve_forever, name="vanguarstew-http", daemon=True) + thread.start() + try: + service.serve_forever() + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) diff --git a/vanguarstew_runtime/state.py b/vanguarstew_runtime/state.py new file mode 100644 index 00000000..a929c620 --- /dev/null +++ b/vanguarstew_runtime/state.py @@ -0,0 +1,305 @@ +"""Durable, private state for the Vanguarstew runtime. + +This database is an operator-local queue, never a public evidence source. It +stores only operational identifiers and private result locations; raw reviewer +output is written to a separate owner-only directory and is not served over +HTTP or included in runtime status responses. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import threading +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + + +def _utcnow() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +def _secure_directory(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + try: + path.chmod(0o700) + except OSError: + pass + + +@dataclass(frozen=True) +class ReviewJob: + """A private unit of pull-request review work.""" + + id: int + delivery_id: str + repository: str + pr_number: int + head_sha: str | None + attempts: int + + +class RuntimeState: + """SQLite-backed queue with atomic claims and owner-only result storage.""" + + def __init__(self, database_path: str | Path, private_result_dir: str | Path): + self.database_path = Path(database_path) + self.private_result_dir = Path(private_result_dir) + _secure_directory(self.database_path.parent) + _secure_directory(self.private_result_dir) + self._lock = threading.RLock() + self._connection = sqlite3.connect( + self.database_path, + timeout=30, + isolation_level=None, + check_same_thread=False, + ) + self._connection.row_factory = sqlite3.Row + self._connection.execute("PRAGMA journal_mode=WAL") + self._connection.execute("PRAGMA foreign_keys=ON") + self._initialize() + try: + self.database_path.chmod(0o600) + except OSError: + pass + + def _initialize(self) -> None: + with self._lock: + self._connection.executescript( + """ + CREATE TABLE IF NOT EXISTS jobs ( + id INTEGER PRIMARY KEY, + delivery_id TEXT NOT NULL UNIQUE, + repository TEXT NOT NULL, + pr_number INTEGER NOT NULL CHECK (pr_number > 0), + head_sha TEXT, + status TEXT NOT NULL CHECK (status IN + ('queued', 'running', 'succeeded', 'deferred', 'failed')), + attempts INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + claimed_at TEXT, + result_path TEXT, + failure_code TEXT + ); + CREATE INDEX IF NOT EXISTS jobs_status_created + ON jobs(status, created_at, id); + CREATE TABLE IF NOT EXISTS runtime_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + """ + ) + + def close(self) -> None: + with self._lock: + self._connection.close() + + def __enter__(self) -> "RuntimeState": + return self + + def __exit__(self, exc_type, exc, traceback) -> None: + self.close() + + def heartbeat(self) -> None: + now = _utcnow() + with self._lock: + self._connection.execute( + """ + INSERT INTO runtime_meta(key, value, updated_at) + VALUES ('heartbeat', ?, ?) + ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at + """, + (now, now), + ) + + def enqueue_pull_request( + self, + *, + delivery_id: str, + repository: str, + pr_number: int, + head_sha: str | None = None, + ) -> bool: + """Add work once. Duplicate delivery ids are deliberately harmless.""" + if not isinstance(delivery_id, str) or not delivery_id.strip(): + raise ValueError("delivery_id must be a non-empty string") + if not isinstance(repository, str) or not repository.strip(): + raise ValueError("repository must be a non-empty string") + if isinstance(pr_number, bool) or not isinstance(pr_number, int) or pr_number <= 0: + raise ValueError("pr_number must be a positive integer") + if head_sha is not None and not isinstance(head_sha, str): + raise ValueError("head_sha must be a string or None") + now = _utcnow() + with self._lock: + cursor = self._connection.execute( + """ + INSERT INTO jobs( + delivery_id, repository, pr_number, head_sha, status, attempts, created_at, updated_at + ) VALUES (?, ?, ?, ?, 'queued', 0, ?, ?) + ON CONFLICT(delivery_id) DO NOTHING + """, + (delivery_id.strip(), repository.strip(), pr_number, head_sha, now, now), + ) + return cursor.rowcount == 1 + + def claim_next(self) -> ReviewJob | None: + """Atomically claim one queued job for this process.""" + now = _utcnow() + with self._lock: + self._connection.execute("BEGIN IMMEDIATE") + try: + row = self._connection.execute( + """ + SELECT id, delivery_id, repository, pr_number, head_sha, attempts + FROM jobs WHERE status = 'queued' ORDER BY created_at, id LIMIT 1 + """ + ).fetchone() + if row is None: + self._connection.execute("COMMIT") + return None + cursor = self._connection.execute( + """ + UPDATE jobs + SET status='running', attempts=attempts+1, claimed_at=?, updated_at=? + WHERE id=? AND status='queued' + """, + (now, now, row["id"]), + ) + self._connection.execute("COMMIT") + except Exception: + self._connection.execute("ROLLBACK") + raise + if cursor.rowcount != 1: + return None + return ReviewJob( + id=int(row["id"]), + delivery_id=str(row["delivery_id"]), + repository=str(row["repository"]), + pr_number=int(row["pr_number"]), + head_sha=row["head_sha"], + attempts=int(row["attempts"]) + 1, + ) + + def defer(self, job_id: int, *, code: str) -> None: + self._transition(job_id, status="deferred", failure_code=code) + + def fail(self, job_id: int, *, code: str) -> None: + self._transition(job_id, status="failed", failure_code=code) + + def complete(self, job_id: int, *, result_path: str) -> None: + self._transition(job_id, status="succeeded", result_path=result_path, failure_code=None) + + def _transition( + self, + job_id: int, + *, + status: str, + result_path: str | None = None, + failure_code: str | None = None, + ) -> None: + if status not in {"succeeded", "deferred", "failed"}: + raise ValueError("invalid terminal job status") + with self._lock: + cursor = self._connection.execute( + """ + UPDATE jobs + SET status=?, result_path=?, failure_code=?, updated_at=? + WHERE id=? AND status='running' + """, + (status, result_path, failure_code, _utcnow(), job_id), + ) + if cursor.rowcount != 1: + raise ValueError("job is not running") + + def requeue_deferred(self) -> int: + """Return safe, policy-deferred work to the queue after an explicit enablement. + + A dry-run installation must not silently discard a signed webhook or + poll result. This method intentionally excludes failed jobs: transport + or model failures need an operator-visible recovery policy, rather than + an unbounded retry loop that can spend money or repeatedly hit GitHub. + """ + with self._lock: + cursor = self._connection.execute( + """ + UPDATE jobs + SET status='queued', claimed_at=NULL, failure_code=NULL, updated_at=? + WHERE status='deferred' AND failure_code IN + ('dry-run', 'inference-not-explicitly-enabled') + """, + (_utcnow(),), + ) + return cursor.rowcount + + def recover_expired_claims(self, *, lease_seconds: int = 900) -> int: + """Requeue work left running by a stopped process after its lease expires.""" + if isinstance(lease_seconds, bool) or not isinstance(lease_seconds, int) or lease_seconds < 1: + raise ValueError("lease_seconds must be a positive integer") + cutoff = (datetime.now(timezone.utc) - timedelta(seconds=lease_seconds)).replace( + microsecond=0 + ).isoformat() + with self._lock: + cursor = self._connection.execute( + """ + UPDATE jobs + SET status='queued', claimed_at=NULL, failure_code='claim-lease-expired', updated_at=? + WHERE status='running' AND claimed_at IS NOT NULL AND claimed_at <= ? + """, + (_utcnow(), cutoff), + ) + return cursor.rowcount + + def write_private_result(self, job_id: int, result: dict[str, Any]) -> str: + """Persist a review result locally with owner-only permissions. + + The returned relative path is an opaque local reference. No public API + resolves it, and callers must not put it in GitHub comments, receipts, + or benchmark artifacts. + """ + if isinstance(job_id, bool) or not isinstance(job_id, int) or job_id <= 0: + raise ValueError("job_id must be a positive integer") + encoded = json.dumps(result, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + target = self.private_result_dir / f"review-{job_id}.json" + temporary = self.private_result_dir / f".review-{job_id}.{os.getpid()}.tmp" + flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC + descriptor = os.open(temporary, flags, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(encoded) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + try: + target.chmod(0o600) + except OSError: + pass + except Exception: + try: + temporary.unlink(missing_ok=True) + except OSError: + pass + raise + return target.name + + def queue_counts(self) -> dict[str, int]: + """Return aggregate operational counts; no repository or PR data escapes.""" + with self._lock: + rows = self._connection.execute( + "SELECT status, COUNT(*) AS total FROM jobs GROUP BY status" + ).fetchall() + counts = {"queued": 0, "running": 0, "succeeded": 0, "deferred": 0, "failed": 0} + counts.update({str(row["status"]): int(row["total"]) for row in rows}) + return counts + + def heartbeat_at(self) -> str | None: + with self._lock: + row = self._connection.execute( + "SELECT value FROM runtime_meta WHERE key='heartbeat'" + ).fetchone() + return None if row is None else str(row["value"])