Implement one-instruction contract compiler

CHANGELOG.md

@@ -4,6 +4,15 @@ All notable changes are documented here. The project follows Semantic Versioning
 
 ## [Unreleased]
 
+### Core
+
+- Added the one-instruction local-project path: bounded repository discovery, typed
+  contract compilation, independent criticism, risk/confidence and authority gates,
+  pre-mutation SHA-256 locking, clarification/resume, Receipt binding, and visible
+  contract provenance in the task UI.
+- Moved manual criteria, checks, artifacts, capabilities, and budgets into an Advanced
+  panel while preserving them as immutable overrides when supplied.
+
 ### Security
 
 - Pinned patched `brace-expansion` releases across transitive dependency trees.

README.md

@@ -144,6 +144,10 @@ and residual risks.
 
 ## What is implemented
 
+- One-instruction local-repository handoff: bounded read-only discovery, a typed
+  compiler plus independent critic, a content-addressed contract locked before the
+  first mutation, and an explicit pause when risk, confidence, or authority is not
+  safe to infer.
 - ReAct planning with independent executor/verifier provider selection and fallback.
 - Strict acceptance contracts with required final artifacts, independent check
   snapshots, baseline regression checks, criterion-to-evidence mappings, Receipt

apps/api/alembic/versions/0009_contract_draft.py

@@ -0,0 +1,37 @@
+"""Add the pre-mutation repository contract draft.
+
+Revision ID: 0009_contract_draft
+Revises: 0008_verified_completion
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0009_contract_draft"
+down_revision: str | None = "0008_verified_completion"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("tasks", sa.Column("contract_draft", sa.JSON(), nullable=True))
+    op.add_column("tasks", sa.Column("contract_hash", sa.String(length=64), nullable=True))
+    op.add_column(
+        "tasks",
+        sa.Column(
+            "contract_status",
+            sa.String(length=20),
+            nullable=False,
+            server_default="not_required",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("tasks", "contract_status")
+    op.drop_column("tasks", "contract_hash")
+    op.drop_column("tasks", "contract_draft")

apps/api/app/db/models/task.py

@@ -38,6 +38,9 @@ class TaskModel(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     baseline_checks: Mapped[list[dict[str, Any]]] = mapped_column(
         JSON, nullable=False, default=list
     )
+    contract_draft: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True)
+    contract_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    contract_status: Mapped[str] = mapped_column(String(20), nullable=False, default="not_required")
     pending_question: Mapped[str | None] = mapped_column(Text, nullable=True)
     authority_schema: Mapped[str] = mapped_column(
         String(40), nullable=False, default="loop.capabilities/v1"

apps/api/app/schemas/contract.py

@@ -0,0 +1,106 @@
+"""Typed, content-addressed acceptance contract for repository tasks."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from app.domain.capability import Capability
+
+
+class ContractCheck(BaseModel):
+    id: str = Field(min_length=1, max_length=80)
+    kind: Literal["command", "file_exists", "file_contains"]
+    command: str | None = Field(default=None, max_length=1_000)
+    path: str | None = Field(default=None, max_length=500)
+    text: str | None = Field(default=None, max_length=2_000)
+    expect_exit: int = 0
+    expect_stdout: str | None = Field(default=None, max_length=2_000)
+    criterion_ids: list[str] = Field(default_factory=list, max_length=12)
+    source: Literal["contract", "system"] = "contract"
+
+    @field_validator("path")
+    @classmethod
+    def validate_path(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        path = value.strip()
+        parts = path.split("/")
+        if "\\" in path or any(part in {"", ".", ".."} for part in parts):
+            raise ValueError("contract paths must be workspace-relative POSIX paths")
+        return path
+
+    @model_validator(mode="after")
+    def validate_target(self) -> ContractCheck:
+        if self.kind == "command" and not (self.command or "").strip():
+            raise ValueError("command checks require command")
+        if self.kind in {"file_exists", "file_contains"} and not self.path:
+            raise ValueError("file checks require path")
+        if self.kind == "file_contains" and not self.text:
+            raise ValueError("file_contains checks require text")
+        return self
+
+
+class RepositoryDiscovery(BaseModel):
+    manifests: list[str] = Field(default_factory=list, max_length=100)
+    scripts: dict[str, str] = Field(default_factory=dict)
+    test_files: list[str] = Field(default_factory=list, max_length=100)
+    build_outputs: list[str] = Field(default_factory=list, max_length=50)
+    quality_checks: list[ContractCheck] = Field(default_factory=list, max_length=16)
+    files_scanned: int = Field(default=0, ge=0)
+    truncated: bool = False
+
+
+class ContractCritique(BaseModel):
+    accepted: bool
+    issues: list[str] = Field(default_factory=list, max_length=12)
+    question: str | None = Field(default=None, max_length=1_000)
+    provider: str = Field(default="unknown", max_length=80)
+    model: str = Field(default="unknown", max_length=160)
+
+
+class ContractModelIdentity(BaseModel):
+    provider: str = Field(default="unknown", max_length=80)
+    model: str = Field(default="unknown", max_length=160)
+
+
+class ContractProposal(BaseModel):
+    criteria: list[str] = Field(min_length=1, max_length=12)
+    checks: list[ContractCheck] = Field(default_factory=list, max_length=72)
+    artifacts: list[str] = Field(default_factory=list, max_length=48)
+    risk: Literal["low", "medium", "high"]
+    assumptions: list[str] = Field(default_factory=list, max_length=12)
+    confidence: int = Field(ge=0, le=100)
+    authority_requests: list[Capability] = Field(default_factory=list, max_length=15)
+
+    @field_validator("criteria", "assumptions")
+    @classmethod
+    def normalize_text_list(cls, value: list[str]) -> list[str]:
+        return list(dict.fromkeys(item.strip() for item in value if item.strip()))
+
+    @field_validator("artifacts")
+    @classmethod
+    def validate_artifacts(cls, value: list[str]) -> list[str]:
+        artifacts = list(dict.fromkeys(item.strip() for item in value if item.strip()))
+        for artifact in artifacts:
+            parts = artifact.split("/")
+            if "\\" in artifact or any(part in {"", ".", ".."} for part in parts):
+                raise ValueError("artifacts must be workspace-relative POSIX paths")
+        return artifacts
+
+    @model_validator(mode="after")
+    def validate_criteria(self) -> ContractProposal:
+        if not self.criteria:
+            raise ValueError("at least one non-empty criterion is required")
+        return self
+
+
+class ContractDraft(ContractProposal):
+    checks: list[ContractCheck] = Field(default_factory=list, max_length=96)
+    artifacts: list[str] = Field(default_factory=list, max_length=64)
+    schema_version: Literal["loop.contract-draft/v1"] = "loop.contract-draft/v1"
+    compiler: ContractModelIdentity
+    discovery: RepositoryDiscovery
+    clarifications: list[str] = Field(default_factory=list, max_length=12)
+    critique: ContractCritique

apps/api/app/schemas/task.py

@@ -15,6 +15,7 @@
 from app.domain.authority_token import AuthorityTokenError, normalize_hosts
 from app.domain.capability import CAPABILITY_SCHEMA_VERSION, Capability
 from app.domain.task import StopReason, TaskStatus
+from app.schemas.contract import ContractDraft
 from app.schemas.file import FileEntry
 from app.schemas.step import LedgerStatus, StepRead
 
@@ -119,8 +120,6 @@ def validate_network_authority(self) -> TaskCreate:
                 self.egress_hosts = sorted(normalize_hosts(self.egress_hosts))
             except AuthorityTokenError as exc:
                 raise ValueError(str(exc)) from exc
-        if self.project_path and self.verification_mode != "judgment" and not self.success_criteria:
-            raise ValueError("strict local-project tasks require user-confirmed success_criteria")
         return self
 
 
@@ -198,6 +197,9 @@ class TaskRead(BaseModel):
     verification_mode: str
     required_checks: list[dict[str, object]]
     baseline_checks: list[dict[str, object]]
+    contract: ContractDraft | None
+    contract_hash: str | None
+    contract_status: str
     pending_question: str | None
     allowed_tools: list[str] | None
     authority: AuthorityRead
@@ -247,6 +249,13 @@ def from_model(cls, model: object) -> TaskRead:
             verification_mode=m.verification_mode,  # type: ignore[attr-defined]
             required_checks=m.required_checks or [],  # type: ignore[attr-defined]
             baseline_checks=m.baseline_checks or [],  # type: ignore[attr-defined]
+            contract=(
+                ContractDraft.model_validate(m.contract_draft)  # type: ignore[attr-defined]
+                if m.contract_draft  # type: ignore[attr-defined]
+                else None
+            ),
+            contract_hash=m.contract_hash,  # type: ignore[attr-defined]
+            contract_status=m.contract_status,  # type: ignore[attr-defined]
             pending_question=m.pending_question,  # type: ignore[attr-defined]
             allowed_tools=m.allowed_tools,  # type: ignore[attr-defined]
             authority=AuthorityRead(

apps/api/app/services/agent_react.py

@@ -55,6 +55,7 @@
 from app.domain.task import StopReason, TaskStatus
 from app.repositories.step import StepRepository
 from app.repositories.task import TaskRepository
+from app.schemas.contract import ContractDraft
 from app.services.completion import (
     attach_baseline,
     completion_gates_pass,
@@ -64,6 +65,12 @@
     merge_completion_checks,
     regressions,
 )
+from app.services.contract import (
+    compile_project_contract,
+    failed_contract_draft,
+    lock_user_project_contract,
+    verify_contract_hash,
+)
 from app.services.ledger import genesis_hash, step_hash
 from app.services.memory import MemoryStore, scoped_memory_root
 from app.services.progress import HistoryEntry, ProgressGuard, compact_history
@@ -257,7 +264,6 @@ async def run(self, task_id: uuid.UUID) -> None:
         self.memory = MemoryStore(
             scoped_memory_root(Path(settings.agent_memory_root), task.owner_id, task.project_id)
         )
-
         # Load the signed skill (if any) BEFORE anything runs. A skill that can't
         # be verified is refused outright — provenance is not optional.
         skill_capabilities: frozenset[Capability] | None = None
@@ -337,6 +343,14 @@ async def run(self, task_id: uuid.UUID) -> None:
             egress_hosts=task.egress_hosts,
         )
         task.resolved_capabilities = sorted_capabilities(resolved_capabilities)
+        if task.project_base_commit:
+            task.workspace_path = str(workspace.root)
+            if not await self._prepare_project_contract(
+                task,
+                workspace,
+                granted_capabilities=set(resolved_capabilities),
+            ):
+                return
         browser_capabilities = {Capability.NET_BROWSER}
         email_capabilities = {Capability.EMAIL_READ, Capability.EMAIL_SEND}
         calendar_capabilities = {Capability.CALENDAR_READ, Capability.CALENDAR_WRITE}
@@ -950,6 +964,107 @@ async def collect_authority_audit(tool: str, _args: dict[str, Any], _result: Any
             if provider_gateway is not None:
                 await provider_gateway.stop()
 
+    async def _prepare_project_contract(
+        self,
+        task: TaskModel,
+        workspace: Workspace,
+        *,
+        granted_capabilities: set[Capability],
+    ) -> bool:
+        """Lock one acceptance source before the executor can mutate the clone."""
+        if task.contract_status == "locked":
+            try:
+                draft = ContractDraft.model_validate(task.contract_draft)
+            except (TypeError, ValueError):
+                draft = None
+            if (
+                draft is None
+                or not task.contract_hash
+                or not draft.critique.accepted
+                or not verify_contract_hash(draft, task.contract_hash)
+            ):
+                task.status = TaskStatus.FAILED.value
+                task.stop_reason = StopReason.ERROR.value
+                task.error = "The locked acceptance contract failed its content hash check."
+                self._ensure_unverified_receipt(task)
+                await self._commit()
+                return False
+            self._apply_locked_contract(task, draft)
+            return True
+
+        if task.criteria_source == "user" and task.rubric:
+            draft, contract_hash = lock_user_project_contract(
+                root=workspace.root,
+                criteria=list(task.rubric),
+                required_checks=list(task.required_checks or []),
+            )
+            task.contract_draft = draft.model_dump(mode="json")
+            task.contract_hash = contract_hash
+            task.contract_status = "locked"
+            self._apply_locked_contract(task, draft)
+            await self._commit()
+            return True
+
+        previous = task.contract_draft if isinstance(task.contract_draft, dict) else {}
+        clarifications = [
+            str(item) for item in previous.get("clarifications", []) if str(item).strip()
+        ]
+        compilation_budget = max(
+            0,
+            task.token_budget - task.tokens_used - _verification_reserve(task.token_budget),
+        )
+        try:
+            compiled = await compile_project_contract(
+                goal=task.goal,
+                root=workspace.root,
+                compiler=self.llm,
+                critic=self.verifier_llm,
+                granted_capabilities=granted_capabilities,
+                required_checks=list(task.required_checks or []),
+                clarifications=clarifications,
+                token_budget=compilation_budget,
+            )
+        except LLMError as exc:
+            task.tokens_used += exc.tokens_spent
+            draft = failed_contract_draft(
+                goal=task.goal,
+                root=workspace.root,
+                issue=f"The contract compiler could not complete: {exc}",
+                clarifications=clarifications,
+            )
+            task.contract_draft = draft.model_dump(mode="json")
+            task.contract_hash = None
+            task.contract_status = "awaiting_input"
+            task.pending_question = draft.critique.question
+            task.status = TaskStatus.AWAITING_INPUT.value
+            await self._commit()
+            return False
+
+        task.tokens_used += compiled.tokens_spent
+        task.contract_draft = compiled.draft.model_dump(mode="json")
+        task.contract_hash = compiled.contract_hash
+        if compiled.contract_hash is None:
+            task.contract_status = "awaiting_input"
+            task.pending_question = compiled.draft.critique.question
+            task.status = TaskStatus.AWAITING_INPUT.value
+            await self._commit()
+            return False
+
+        task.contract_status = "locked"
+        task.criteria_source = "compiled"
+        task.pending_question = None
+        self._apply_locked_contract(task, compiled.draft)
+        await self._commit()
+        return True
+
+    @staticmethod
+    def _apply_locked_contract(task: TaskModel, draft: ContractDraft) -> None:
+        task.rubric = list(draft.criteria)
+        task.verification_mode = "strict"
+        task.required_checks = [
+            check.model_dump(mode="json", exclude_none=True) for check in draft.checks
+        ]
+
     def _resolve_sandbox(self) -> tuple[str | None, str, str | None]:
         """(image, label): which sandbox to use for run_command, and how to label
         it. 'container' jails commands in Docker; 'inline' runs on the host (a