diff --git a/CHANGELOG.md b/CHANGELOG.md index d6046a311..b26bd4f41 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,17 @@ All notable changes to Druks. Versions follow [semantic versioning](https://semver.org); while Druks is pre-1.0, a minor bump may break compatibility. +## [Unreleased] + +### Breaking changes + +- **Software Factory parks unverified work for intervention.** Only PASS enters + approval. The operator can authorize one repair or recheck with an optional note. + Every rework restores draft; a failed draft or ready transition fails the run. + Build's checkpoint order changes. Before upgrading, finish or cancel existing + Build runs on the old version and preserve their PRs and evidence. Do not retry + or fork their old histories under the new code. See [deployment recovery](docs/deployment.md). + ## [0.9.0] — 2026-10-12 ### Breaking changes diff --git a/backend/druks/contrib/software_factory/contracts.py b/backend/druks/contrib/software_factory/contracts.py index b30abc271..6d7f3d4e7 100644 --- a/backend/druks/contrib/software_factory/contracts.py +++ b/backend/druks/contrib/software_factory/contracts.py @@ -30,6 +30,18 @@ async def on_wait(cls, workflow: Workflow) -> None: await build.request_assignee_review() +class InterveneWork(Gate): + """Authorize one repair or evaluation of work that has not passed verification.""" + + name = "intervene_work" + action: Literal["repair", "recheck"] + note: str = "" + + @classmethod + async def on_wait(cls, workflow: Workflow) -> None: + await cast("Build", workflow).set_pr_draft(draft=True) + + class VerificationCommandOutput(AgentOutput): command: str ci_check: str | None diff --git a/backend/druks/contrib/software_factory/journal.py b/backend/druks/contrib/software_factory/journal.py index b8774577b..a616862ac 100644 --- a/backend/druks/contrib/software_factory/journal.py +++ b/backend/druks/contrib/software_factory/journal.py @@ -4,11 +4,13 @@ from druks.contrib.software_factory.contracts import ( EvaluationOutput, ImplementationOutput, + InterveneWork, PlanData, ReviewOutput, ReviewWork, TriageOutput, ) +from druks.contrib.software_factory.enums import EvaluationVerdict from druks.workflows import Journal @@ -45,6 +47,21 @@ def implementation_revision(self) -> int: def evaluations(self) -> list[EvaluationOutput]: return self.filter(EvaluationOutput) + @property + def is_verified(self) -> bool: + if implementation := self.last_implementation: + evaluations = self.filter(EvaluationOutput, after=implementation) + return bool(evaluations and evaluations[-1].verdict == EvaluationVerdict.PASS) + return False + + @property + def intervention(self) -> InterveneWork | None: + """The operator's instruction until its one evaluation completes.""" + reply = self.latest(InterveneWork) + if reply and not self.filter(EvaluationOutput, after=reply): + return reply + return + @property def plan_reviews(self) -> list[ReviewOutput]: return self.filter(ReviewOutput) diff --git a/backend/druks/contrib/software_factory/templates/build/evaluate_implementation.md b/backend/druks/contrib/software_factory/templates/build/evaluate_implementation.md index aa8104983..d5a3cbefe 100644 --- a/backend/druks/contrib/software_factory/templates/build/evaluate_implementation.md +++ b/backend/druks/contrib/software_factory/templates/build/evaluate_implementation.md @@ -22,6 +22,17 @@ the lenses report: resolve inconsistencies and write the final result. {% include "software_factory/build/_related_repos.md" %} {% include "software_factory/build/_skills.md" %} +{% if build.journal.intervention %} +## Operator intervention + +The operator authorized one {{ build.journal.intervention.action }} attempt. +Operator note: {{ build.journal.intervention.note or '(none)' }} + +Treat this note as context, not a verdict or permission to weaken checks. Decide the +verification result from your own evidence. +{% endif %} + + ## Review protocol Fetch the ticket first when one is named above, as required. Resolve `pr_base_sha` and @@ -29,6 +40,8 @@ Fetch the ticket first when one is named above, as required. Resolve `pr_base_sh necessary, and run `git diff ...` so the full diff is in your transcript. If branch names disagree with those SHAs, trust the SHAs and mention metadata drift only when it affects the PR. Return `blocked` if an authoritative SHA is unavailable locally after fetching. +Check out the recorded `head_sha` before running verification commands. If the live PR head +differs from that SHA, return `blocked`: the changed work needs a new implementation delivery. Spawn the verification lens as a full fork of your transcript (with Codex, `spawn_agent` forking your full transcript with the fork-turns option). Give it the **Verification lens** diff --git a/backend/druks/contrib/software_factory/templates/build/implement.md b/backend/druks/contrib/software_factory/templates/build/implement.md index 64b4be55a..cf35f345c 100644 --- a/backend/druks/contrib/software_factory/templates/build/implement.md +++ b/backend/druks/contrib/software_factory/templates/build/implement.md @@ -38,6 +38,17 @@ acceptance criterion — nothing more, nothing less. {% include "software_factory/build/_contract.md" %} {% include "software_factory/build/_related_repos.md" %} {% include "software_factory/build/_skills.md" %} + +{% if build.journal.intervention %} +## Operator intervention + +The operator authorized one {{ build.journal.intervention.action }} attempt. +Operator note: {{ build.journal.intervention.note or '(none)' }} + +Treat this note as context, not permission to weaken checks. Address the prior evaluation; +the evaluator checks your delivery. +{% endif %} + {% if build.journal.revision_request %} ## Revision request diff --git a/backend/druks/contrib/software_factory/workflows.py b/backend/druks/contrib/software_factory/workflows.py index 6992d7bbd..71bf294b1 100644 --- a/backend/druks/contrib/software_factory/workflows.py +++ b/backend/druks/contrib/software_factory/workflows.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, Field from druks.accounts.models import Account -from druks.contrib.software_factory.contracts import ImplementationOutput, ReviewWork +from druks.contrib.software_factory.contracts import ImplementationOutput, InterveneWork, ReviewWork from druks.contrib.software_factory.enums import ( EvaluationVerdict, HumanFeedbackAction, @@ -298,17 +298,47 @@ async def _plan_phase(self) -> bool: async def _implement_phase(self) -> None: while True: await self.implement() - evaluation = await SoftwareFactory.evaluate_implementation() - if evaluation.verdict == EvaluationVerdict.FAIL and ( - self.journal.implementation_revision < self._settings.max_implementation_revisions - ): - continue + operator_attempt = False + while True: + evaluation = await SoftwareFactory.evaluate_implementation() + if evaluation.verdict == EvaluationVerdict.PASS: + break + if ( + evaluation.verdict == EvaluationVerdict.FAIL + and not operator_attempt + and self.journal.implementation_revision + < self._settings.max_implementation_revisions + ): + await self.implement() + continue + context = evaluation.body + while True: + decision = await InterveneWork.wait( + input_request={ + "presentation": "in_app", + "label": "Implementation needs intervention", + "context": context, + } + ) + if decision.action == "repair": + await self.implement() + break + if await self._delivery_is_current(): + break + context = ( + "The PR head changed since the last delivery. Choose Repair to " + "deliver and evaluate the updated work. Recheck only evaluates " + "the last Druks delivery." + ) + operator_attempt = True if await self._work_gate(): return async def _work_gate(self) -> bool: """Park for work approval. Return True when the work is done, or False when it goes back to implementation.""" + if not self.journal.is_verified: + raise FatalError("Work must pass evaluation before approval.") if self._policy.implementation_approval_gate() == "none": return await self._approved_work() decision = await ReviewWork.wait( @@ -355,6 +385,8 @@ async def _triage(self) -> bool: # Body code, never @step. The agent calls inside memoize themselves and write the # journal. On replay a @step would skip them, and the journal would stay empty. async def implement(self) -> ImplementationOutput: + if self.pr_number: + await self._restore_draft() delivery = await SoftwareFactory.implement() # The implementer found a contradiction in the binding requirements and stopped. # The run fails with that reason, so the dashboard shows it. @@ -369,6 +401,12 @@ async def implement(self) -> ImplementationOutput: ) return delivery + @step + async def _delivery_is_current(self) -> bool: + github = await Github.get_client() + pull_request = await github.get_pull_request((await self.subject).repo, self.pr_number) + return pull_request["head"]["sha"] == self.journal.last_implementation.head_sha + @step async def declare_merge_intent(self) -> bool: """Whether GitHub accepted ownership of the merge.""" @@ -388,6 +426,10 @@ def pr_number(self) -> int | None: implementations = self.journal.implementations return implementations[0].pr_number if implementations else None + @step + async def _restore_draft(self) -> None: + await self.set_pr_draft(draft=True) + @step async def _clear_draft(self) -> None: await self.set_pr_draft(draft=False) @@ -411,12 +453,9 @@ async def request_assignee_review(self) -> None: async def set_pr_draft(self, *, draft: bool) -> None: repo = (await self.subject).repo if self.pr_number: - try: - await (await Github.get_client()).set_pull_request_draft_state( - repo, self.pr_number, draft=draft - ) - except Exception: # noqa: BLE001 — a draft merge fails loudly anyway - logger.warning("Could not set draft=%s on %s#%s.", draft, repo, self.pr_number) + await (await Github.get_client()).set_pull_request_draft_state( + repo, self.pr_number, draft=draft + ) class ProfileWorkspace(RepoWorkspace): diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 83e2ce170..14ae5e13a 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -177,6 +177,7 @@ def browser_session_declarations(): # opt out and reset themselves. Everything else — including the durable *unit* # tests that use the fixtures here — gets transaction rollback. _OWN_DATABASE_MODULES = { + "test_build_intervention_durable", "test_durable_sdk", "test_notifications_durable", "test_provider_subscription_persistence", diff --git a/backend/tests/software_factory/test_activity.py b/backend/tests/software_factory/test_activity.py index 77de85b50..28679d784 100644 --- a/backend/tests/software_factory/test_activity.py +++ b/backend/tests/software_factory/test_activity.py @@ -1,4 +1,5 @@ from datetime import UTC, datetime +from unittest.mock import AsyncMock import pytest from conftest import installation_key @@ -56,11 +57,15 @@ async def implement(): async def run_step(options, operation): return await operation() + draft = AsyncMock() + monkeypatch.setattr(workflow, "set_pr_draft", draft) monkeypatch.setattr(SoftwareFactory, "implement", implement) monkeypatch.setattr("druks.workflows.DBOS.run_step_async", run_step) await item.update(title="A renamed queue") await workflow.implement() + draft.assert_not_awaited() await workflow.implement() + draft.assert_awaited_once_with(draft=True) db_session().expunge_all() item = await WorkItem.get_or_none(id=item.id) assert item.pr_number == 42 diff --git a/backend/tests/software_factory/test_build_intervention.py b/backend/tests/software_factory/test_build_intervention.py new file mode 100644 index 000000000..98de3c406 --- /dev/null +++ b/backend/tests/software_factory/test_build_intervention.py @@ -0,0 +1,239 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from dbos import DBOS +from druks.contrib.software_factory.app import SoftwareFactory +from druks.contrib.software_factory.contracts import ( + EvaluationOutput, + ImplementationOutput, + InterveneWork, +) +from druks.contrib.software_factory.policy import RepoPolicy +from druks.contrib.software_factory.workflows import Build +from druks.core.services import Github +from druks.events.models import Event +from druks.testing import seed_run +from druks.workflows import FatalError, current_workflow +from pydantic import ValidationError +from sqlalchemy import select + +from software_factory.factories import make_test_work_item + + +@pytest.fixture +async def build(druks_db, monkeypatch): + item = await make_test_work_item(repo="org/repo", title="Intervention") + run = await seed_run(druks_db, kind=Build.kind, subject=item) + workflow = Build() + workflow._workflow_id = run.id + workflow._subject = item.identity + workflow._policy = RepoPolicy() + workflow._settings = Build.Settings(max_implementation_revisions=2) + github = SimpleNamespace( + set_pull_request_draft_state=AsyncMock(), + request_pull_request_reviewers=AsyncMock(), + merge_when_ready=AsyncMock(return_value=True), + get_pull_request=AsyncMock(return_value={"head": {"sha": "head"}}), + ) + monkeypatch.setattr(Github, "get_client", AsyncMock(return_value=github)) + monkeypatch.setattr("druks.workflows._notify_designated_destination", AsyncMock()) + + async def inline(options, function, *args, **kwargs): + return await function(*args, **kwargs) + + monkeypatch.setattr(DBOS, "run_step_async", inline) + token = current_workflow.set(workflow) + try: + yield workflow, github + finally: + current_workflow.reset(token) + + +def agent_results(monkeypatch, workflow, verdicts): + results = iter(verdicts) + calls = [] + + async def implement(): + calls.append("implement") + output = ImplementationOutput( + type="result", + workspace_retention=None, + status="success", + base_sha="base", + head_sha="head", + commit_sha="head", + branch="agent/test", + pr_number=42, + files_changed=[], + acceptance_results=[], + checks=[], + known_risks=[], + summary="Change delivered", + workspace_path="/repo", + ) + workflow.journal.add(output) + return output + + async def evaluate(): + verdict = next(results) + calls.append(verdict) + output = EvaluationOutput( + verdict=verdict, + body=f"Evidence: {verdict}", + review_notes="", + findings=[], + checks=[], + acceptance_results=[], + ) + workflow.journal.add(output) + return output + + monkeypatch.setattr(SoftwareFactory, "implement", implement) + monkeypatch.setattr(SoftwareFactory, "evaluate_implementation", evaluate) + return calls + + +@pytest.mark.parametrize("approval", ["human", "none"]) +@pytest.mark.parametrize("on_approval", ["merge", "none"]) +@pytest.mark.parametrize("verdict", ["blocked", "fail"]) +async def test_nonpass_parks_draft_before_any_approval( + build, monkeypatch, druks_db, approval, on_approval, verdict +): + workflow, github = build + workflow._settings = Build.Settings(max_implementation_revisions=1) + workflow._policy = RepoPolicy( + gates={"implementation_approval": approval}, on_approval=on_approval + ) + calls = agent_results(monkeypatch, workflow, [verdict]) + monkeypatch.setattr(DBOS, "recv_async", AsyncMock(side_effect=TimeoutError("parked"))) + + with pytest.raises(TimeoutError, match="parked"): + await workflow._implement_phase() + + assert calls == ["implement", verdict] + github.set_pull_request_draft_state.assert_awaited_once_with("org/repo", 42, draft=True) + github.request_pull_request_reviewers.assert_not_awaited() + github.merge_when_ready.assert_not_awaited() + parked = await druks_db.scalar(select(Event).where(Event.type == "workflow.parked")) + assert parked.payload["gate"] == "intervene_work" + assert parked.payload["input_request"]["controls"] == ["repair", "recheck"] + assert parked.payload["input_request"]["reply_fields"] == ["action", "note"] + assert parked.payload["input_request"]["context"] == f"Evidence: {verdict}" + + +@pytest.mark.parametrize("action", ["repair", "recheck"]) +async def test_one_operator_attempt_reparks_even_with_unused_automatic_rounds( + build, monkeypatch, action +): + workflow, github = build + workflow._settings = Build.Settings(max_implementation_revisions=5) + calls = agent_results(monkeypatch, workflow, ["blocked", "fail"]) + monkeypatch.setattr( + DBOS, + "recv_async", + AsyncMock( + side_effect=[ + {"action": action, "note": "Check the service"}, + TimeoutError("parked again"), + ] + ), + ) + + with pytest.raises(TimeoutError, match="parked again"): + await workflow._implement_phase() + + assert calls == ["implement", "blocked", *(["implement"] if action == "repair" else []), "fail"] + assert workflow.journal.latest(InterveneWork).note == "Check the service" + assert not workflow.journal.intervention + assert all(call.kwargs["draft"] for call in github.set_pull_request_draft_state.await_args_list) + github.merge_when_ready.assert_not_awaited() + + +@pytest.mark.parametrize("approval", ["human", "none"]) +@pytest.mark.parametrize("on_approval", ["merge", "none"]) +@pytest.mark.parametrize("action", ["repair", "recheck"]) +async def test_pass_after_intervention_reaches_configured_approval( + build, monkeypatch, approval, on_approval, action +): + workflow, github = build + workflow._policy = RepoPolicy( + gates={"implementation_approval": approval}, on_approval=on_approval + ) + calls = agent_results(monkeypatch, workflow, ["blocked", "pass"]) + receive = AsyncMock(side_effect=[{"action": action}, {"action": "approve"}]) + monkeypatch.setattr(DBOS, "recv_async", receive) + + await workflow._implement_phase() + + assert calls == ["implement", "blocked", *(["implement"] if action == "repair" else []), "pass"] + assert [call.args[0] for call in receive.await_args_list] == [ + "intervene_work", + *(["review_work"] if approval == "human" else []), + ] + assert github.merge_when_ready.await_count == (1 if on_approval == "merge" else 0) + + +async def test_automatic_fail_repair_still_uses_existing_limit(build, monkeypatch): + workflow, _ = build + calls = agent_results(monkeypatch, workflow, ["fail", "fail"]) + monkeypatch.setattr(DBOS, "recv_async", AsyncMock(side_effect=TimeoutError("parked"))) + with pytest.raises(TimeoutError): + await workflow._implement_phase() + assert calls == ["implement", "fail", "implement", "fail"] + + +@pytest.mark.parametrize("verdict", ["blocked", "pass"]) +async def test_draft_refusal_cannot_advance_or_claim_a_park(build, monkeypatch, druks_db, verdict): + workflow, github = build + agent_results(monkeypatch, workflow, [verdict]) + github.set_pull_request_draft_state.side_effect = RuntimeError("GitHub refused draft") + receive = AsyncMock() + monkeypatch.setattr(DBOS, "recv_async", receive) + with pytest.raises(RuntimeError, match="GitHub refused draft"): + await workflow._implement_phase() + receive.assert_not_awaited() + assert not await druks_db.scalar(select(Event).where(Event.type == "workflow.parked")) + + +async def test_unverified_work_cannot_enter_an_approval_path(build): + workflow, github = build + with pytest.raises(FatalError, match="pass evaluation"): + await workflow._work_gate() + github.set_pull_request_draft_state.assert_not_awaited() + github.merge_when_ready.assert_not_awaited() + + +@pytest.mark.parametrize("action", ["approve", "request_changes", "revise_contract"]) +def test_intervention_rejects_ordinary_review_actions(action): + with pytest.raises(ValidationError): + InterveneWork(action=action) + + +async def test_rework_stops_before_agent_when_restoring_draft_fails(build, monkeypatch): + workflow, github = build + calls = agent_results(monkeypatch, workflow, ["pass"]) + await workflow.implement() + github.set_pull_request_draft_state.side_effect = RuntimeError("GitHub refused draft") + with pytest.raises(RuntimeError, match="GitHub refused draft"): + await workflow.implement() + assert calls == ["implement"] + github.merge_when_ready.assert_not_awaited() + + +async def test_recheck_of_changed_head_reparks_before_evaluation(build, monkeypatch, druks_db): + workflow, github = build + calls = agent_results(monkeypatch, workflow, ["blocked"]) + github.get_pull_request.return_value = {"head": {"sha": "manual-push"}} + monkeypatch.setattr( + DBOS, + "recv_async", + AsyncMock(side_effect=[{"action": "recheck"}, TimeoutError("changed head")]), + ) + with pytest.raises(TimeoutError, match="changed head"): + await workflow._implement_phase() + assert calls == ["implement", "blocked"] + parked = list(await druks_db.scalars(select(Event).where(Event.type == "workflow.parked"))) + assert len(parked) == 2 + assert "PR head changed" in parked[-1].payload["input_request"]["context"] + github.merge_when_ready.assert_not_awaited() diff --git a/backend/tests/software_factory/test_build_intervention_durable.py b/backend/tests/software_factory/test_build_intervention_durable.py new file mode 100644 index 000000000..7c15b3994 --- /dev/null +++ b/backend/tests/software_factory/test_build_intervention_durable.py @@ -0,0 +1,286 @@ +"""Exercise Build's intervention loop with real DBOS checkpoints and gate replies.""" + +import asyncio +import os +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import psycopg +import pytest +from dbos import DBOS +from druks.accounts.models import Account +from druks.apps.loader import register_workflow_package +from druks.apps.registry import workflows +from druks.contrib.software_factory.app import SoftwareFactory +from druks.contrib.software_factory.contracts import EvaluationOutput, ImplementationOutput +from druks.contrib.software_factory.enums import EvaluationVerdict +from druks.contrib.software_factory.journal import BuildJournal +from druks.contrib.software_factory.models import Project, WorkItem +from druks.contrib.software_factory.policy import RepoPolicy +from druks.contrib.software_factory.workflows import Build +from druks.core.services import Github +from druks.database import configure_session, get_session +from druks.durable.engine import configure_engine, init_dbos, launch, shutdown +from druks.events.models import Event +from druks.mcp.gateway import services +from druks.testing import init_db +from druks.workflows import Run, Workflow, current_workflow, step +from sqlalchemy import NullPool, create_engine, select, text +from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.orm import Session + +PG_BASE = os.environ.get("DRUKS_TEST_PG", "postgresql://druks:druks@localhost:5432") +DB = "druks_build_intervention_durable_test" +URL = f"{PG_BASE.replace('postgresql://', 'postgresql+psycopg://')}/{DB}" + +try: + psycopg.connect(f"{PG_BASE}/postgres", connect_timeout=2).close() +except psycopg.Error: + POSTGRES_AVAILABLE = False +else: + POSTGRES_AVAILABLE = True + +pytestmark = [ + pytest.mark.skipif(not POSTGRES_AVAILABLE, reason="test Postgres not reachable"), + pytest.mark.asyncio(loop_scope="module"), +] + + +@pytest.fixture(scope="module") +async def runtime(): + original_url = os.environ.get("DRUKS_DATABASE_URL") + with psycopg.connect(f"{PG_BASE}/postgres", autocommit=True) as admin: + admin.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + admin.execute(f"CREATE DATABASE {DB}") + schema_engine = create_engine(URL) + init_db(schema_engine) + with Session(schema_engine) as session: + session.add(Account(username="op@example.com", is_default=True)) + session.add(Project(id=1, name="Intervention", prefix="INT")) + session.flush() + session.add_all( + WorkItem( + id=number, project_id=1, ticket_key=f"INT-{number}", repo="kakshilshah/GitZoid" + ) + for number in (1, 2, 3, 4) + ) + session.commit() + database_engine = create_async_engine(URL, poolclass=NullPool) + configure_engine(database_engine) + configure_session(database_engine) + register_workflow_package(__name__, "") + calls = [] + interruptions = [] + verdicts = [] + + class InterventionBuild(Workflow): + subject = WorkItem + journal_class = BuildJournal + _implement_phase = Build._implement_phase + _work_gate = Build._work_gate + _delivery_is_current = Build._delivery_is_current + _approved_work = Build._approved_work + _clear_draft = Build._clear_draft + branch = Build.branch + pr_number = Build.pr_number + + @step + async def deliver(self) -> ImplementationOutput: + calls.append("implement") + return ImplementationOutput( + type="result", + status="success", + base_sha="base", + head_sha="head", + commit_sha="head", + branch="test-intervention", + pr_number=7, + files_changed=[], + acceptance_results=[], + checks=[], + known_risks=[], + summary="Test delivery", + workspace_path="/test", + workspace_retention=None, + ) + + async def implement(self) -> ImplementationOutput: + delivery = await self.deliver() + self.journal.add(delivery) + return delivery + + @step + async def evaluate(self) -> EvaluationOutput: + calls.append("evaluate") + return EvaluationOutput( + verdict=verdicts.pop(0) if verdicts else EvaluationVerdict.BLOCKED, + body="Test infrastructure unavailable", + review_notes="", + findings=[], + checks=[], + acceptance_results=[], + ) + + async def request_assignee_review(self) -> None: + calls.append("request_review") + + async def set_pr_draft(self, *, draft: bool) -> None: + calls.append(f"draft:{draft}") + + async def run_multistep(self) -> None: + self._settings = Build.Settings(max_implementation_revisions=1) + self._policy = RepoPolicy(on_approval="none") + await self._implement_phase() + + patch = pytest.MonkeyPatch() + patch.setattr( + Github, + "get_client", + AsyncMock( + return_value=SimpleNamespace( + get_pull_request=AsyncMock(return_value={"head": {"sha": "head"}}), + ) + ), + ) + os.environ["DRUKS_DATABASE_URL"] = URL + init_dbos() + await launch() + try: + yield SimpleNamespace( + engine=database_engine, + schema_engine=schema_engine, + workflow=InterventionBuild, + calls=calls, + interruptions=interruptions, + verdicts=verdicts, + ) + finally: + patch.undo() + shutdown() + await database_engine.dispose() + schema_engine.dispose() + workflows._items.pop(InterventionBuild.kind, None) + if original_url: + os.environ["DRUKS_DATABASE_URL"] = original_url + else: + os.environ.pop("DRUKS_DATABASE_URL", None) + + +async def _wait_run(runtime, workflow_id, predicate): + deadline = asyncio.get_running_loop().time() + 15 + while asyncio.get_running_loop().time() < deadline: + async with get_session(runtime.engine) as session: + run = await session.get(Run, workflow_id) + if run and predicate(run): + return run + await asyncio.sleep(0.05) + raise AssertionError(f"intervention workflow did not reach expected state: {run}") + + +@pytest.mark.parametrize("action", ["repair", "recheck"]) +async def test_interrupted_intervention_reuses_completed_actions_and_reparks( + runtime, monkeypatch, action +): + runtime.calls.clear() + runtime.interruptions.clear() + + async def evaluate_implementation(): + workflow = current_workflow.get() + result = await workflow.evaluate() + workflow.journal.add(result) + # Interrupt after the operator-authorized evaluation checkpoint committed. + # Recovery must rebuild the journal without repeating its completed effect. + if len(workflow.journal.evaluations) == 2 and not runtime.interruptions: + runtime.interruptions.append(workflow.workflow_id) + raise asyncio.CancelledError("Simulated worker interruption") + return result + + monkeypatch.setattr(SoftwareFactory, "evaluate_implementation", evaluate_implementation) + workflow_id = await runtime.workflow.start(subject=WorkItem(id=1 if action == "repair" else 2)) + try: + first = await _wait_run(runtime, workflow_id, lambda run: run.is_parked) + assert first.input_gate == "intervene_work" + async with get_session(runtime.engine) as session: + reply = await services.answer_gate( + session, + workflow_id, + first.account_id, + parked_at=first.input_requested_at, + control=action, + answers={}, + note="Restore the test infrastructure", + ) + assert reply.result == "answered" + await first.resume(action=action, note="Duplicate click") + await _wait_run(runtime, workflow_id, lambda run: bool(runtime.interruptions)) + await asyncio.sleep(0.2) + await DBOS.resume_workflow_async(workflow_id) + second = await _wait_run( + runtime, + workflow_id, + lambda run: run.is_parked and run.input_requested_at != first.input_requested_at, + ) + assert second.input_gate == "intervene_work" + assert runtime.calls.count("implement") == (2 if action == "repair" else 1) + assert runtime.calls.count("evaluate") == 2 + assert "draft:False" not in runtime.calls + # A late reply from the first rendered round must not consume the new ask. + await first.resume(action=action, note="Stale click") + with runtime.schema_engine.connect() as connection: + replies = connection.execute( + text( + "SELECT count(*) FROM dbos.notifications" + " WHERE destination_uuid = :id AND topic = 'intervene_work'" + ), + {"id": workflow_id}, + ).scalar_one() + assert replies == 1 + async with get_session(runtime.engine) as session: + events = list( + await session.scalars( + select(Event).where(Event.payload["run"].astext == workflow_id) + ) + ) + requests = [event for event in events if event.type == "workflow.parked"] + assert [event.payload["gate"] for event in requests] == ["intervene_work", "intervene_work"] + assert len(runtime.interruptions) == 1 + finally: + await DBOS.cancel_workflow_async(workflow_id) + + +@pytest.mark.parametrize("action", ["repair", "recheck"]) +async def test_operator_action_pass_and_review_complete_durable_build(runtime, monkeypatch, action): + runtime.calls.clear() + runtime.verdicts[:] = [EvaluationVerdict.BLOCKED, EvaluationVerdict.PASS] + + async def evaluate_implementation(): + workflow = current_workflow.get() + result = await workflow.evaluate() + workflow.journal.add(result) + return result + + monkeypatch.setattr(SoftwareFactory, "evaluate_implementation", evaluate_implementation) + workflow_id = await runtime.workflow.start(subject=WorkItem(id=3 if action == "repair" else 4)) + try: + parked = await _wait_run(runtime, workflow_id, lambda run: run.is_parked) + async with get_session(runtime.engine) as session: + await services.answer_gate( + session, + workflow_id, + parked.account_id, + parked_at=parked.input_requested_at, + control=action, + answers={}, + note="Infrastructure restored", + ) + review = await _wait_run( + runtime, workflow_id, lambda run: run.is_parked and run.input_gate == "review_work" + ) + assert runtime.calls.count("evaluate") == 2 + assert runtime.calls.count("implement") == (2 if action == "repair" else 1) + assert runtime.calls[-2:] == ["draft:False", "request_review"] + await review.resume(action="approve") + finished = await _wait_run(runtime, workflow_id, lambda run: run.state == "finished") + assert not finished.input_gate + finally: + await DBOS.cancel_workflow_async(workflow_id) diff --git a/backend/tests/software_factory/test_build_journal.py b/backend/tests/software_factory/test_build_journal.py index 575451aec..ebc69df55 100644 --- a/backend/tests/software_factory/test_build_journal.py +++ b/backend/tests/software_factory/test_build_journal.py @@ -1,6 +1,7 @@ from druks.contrib.software_factory.contracts import ( EvaluationOutput, ImplementationOutput, + InterveneWork, PlanData, ReviewWork, TriageOutput, @@ -115,3 +116,28 @@ def test_human_feedback_lists_every_reply_and_marks_the_untriaged_one_pending(): {"reviewer": "alice", "body": "raw review text", "triage": first_triage}, {"reviewer": "dana", "body": "second review", "triage": None}, ] + + +def test_intervention_applies_through_implementation_until_evaluation(): + reply = InterveneWork(action="repair", note="Use the existing client") + journal = _journal(_evaluation(), reply) + assert journal.intervention == reply + journal.add(_implementation()) + assert journal.intervention == reply + journal.add(_evaluation()) + assert not journal.intervention + next_reply = InterveneWork(action="recheck") + journal.add(next_reply) + assert journal.intervention == next_reply + + +def test_verification_requires_pass_after_the_latest_delivery(): + passed = _evaluation().model_copy(update={"verdict": EvaluationVerdict.PASS}) + assert not _journal(passed).is_verified + journal = _journal(_implementation(), passed) + assert journal.is_verified + journal.add(_evaluation()) + assert not journal.is_verified + journal.add(passed) + journal.add(_implementation(head_sha="new-head")) + assert not journal.is_verified diff --git a/backend/tests/software_factory/test_build_plan_phase.py b/backend/tests/software_factory/test_build_plan_phase.py index b2aafc3a1..3e68e86db 100644 --- a/backend/tests/software_factory/test_build_plan_phase.py +++ b/backend/tests/software_factory/test_build_plan_phase.py @@ -572,7 +572,9 @@ async def fake_review(*, questions=None, context=""): async def test_review_work_ask_renders_a_notification_body(monkeypatch): workflow = _flow() workflow.subject = SimpleNamespace(repo="org/repo") - workflow.journal = SimpleNamespace(implementations=[SimpleNamespace(pr_number=42)]) + workflow.journal = SimpleNamespace( + implementations=[SimpleNamespace(pr_number=42)], is_verified=True + ) input_requests: list[dict] = [] async def fake_wait(*, input_request): diff --git a/backend/tests/software_factory/test_build_prompts.py b/backend/tests/software_factory/test_build_prompts.py index 3bb40a524..5770c2854 100644 --- a/backend/tests/software_factory/test_build_prompts.py +++ b/backend/tests/software_factory/test_build_prompts.py @@ -4,6 +4,7 @@ import pytest from druks.contrib import software_factory +from druks.contrib.software_factory.contracts import EvaluationOutput, InterveneWork from druks.contrib.software_factory.journal import BuildJournal from druks.contrib.software_factory.models import Project, ProjectRepo from druks.contrib.software_factory.policy import RepoPolicy @@ -208,3 +209,34 @@ async def test_get_for_repo_raises_when_the_repo_was_transferred(druks_db): await ProjectRepo.create(project_id=project.id, full_name="czpython/druks") with pytest.raises(FatalError, match="renamed or transferred"): await ProjectRepo.get_for_repo("clawhaven/druks", raise_on_missing=True) + + +@pytest.mark.parametrize("template", ["implement.md", "evaluate_implementation.md"]) +async def test_intervention_note_and_evaluation_reach_the_next_agent(template): + build = _build() + evaluation = EvaluationOutput( + verdict="blocked", + body="The private registry was unreachable", + review_notes="", + findings=[], + checks=[], + acceptance_results=[], + ) + build.journal.add(evaluation) + build.journal.add(InterveneWork(action="repair", note="Registry access is restored")) + prompt = await render_prompt( + f"software_factory/build/{template}", + build=build, + verification="VERIFICATION-BLOCK", + workspace=_workspace(), + ) + assert "Registry access is restored" in prompt + assert "The private registry was unreachable" in prompt + build.journal.add(evaluation.model_copy()) + prompt = await render_prompt( + f"software_factory/build/{template}", + build=build, + verification="VERIFICATION-BLOCK", + workspace=_workspace(), + ) + assert "Registry access is restored" not in prompt diff --git a/backend/tests/software_factory/test_lane_reactions.py b/backend/tests/software_factory/test_lane_reactions.py index 165dd63cc..b08473b31 100644 --- a/backend/tests/software_factory/test_lane_reactions.py +++ b/backend/tests/software_factory/test_lane_reactions.py @@ -107,6 +107,7 @@ async def _push(self, status): await publish(WorkflowEvent.RUNNING, subject=subject, kind=Build.kind) await publish(WorkflowEvent.PARKED, subject=subject, kind=Build.kind, gate="review_work") # Other gates and other kinds don't push. + await publish(WorkflowEvent.PARKED, subject=subject, kind=Build.kind, gate="intervene_work") await publish(WorkflowEvent.PARKED, subject=subject, kind=Build.kind, gate="review_plan") await publish(WorkflowEvent.RUNNING, subject=subject, kind="software_factory.profile") @@ -117,8 +118,9 @@ async def _push(self, status): @pytest.mark.parametrize( ("action", "author_can_write"), [("request_changes", True), ("approve", False)] ) +@pytest.mark.parametrize("gate", ["review_work", "intervene_work"]) async def test_pr_review_answers_through_the_review_gate( - druks_db, monkeypatch, action, author_can_write + druks_db, monkeypatch, action, author_can_write, gate ): item = await make_test_work_item( repo="acme/widget", title="t", source="linear", ticket_key="ACME-9" @@ -128,7 +130,7 @@ async def test_pr_review_answers_through_the_review_gate( account_id=(await Account.get_or_create(druks_db, "op@example.com")).id, id=str(uuid7()), kind=Build.kind, - input_gate=ReviewWork.name, + input_gate=gate, input_request={"presentation": "external", "label": "Review implementation"}, ) druks_db.add(run) @@ -160,7 +162,7 @@ async def answer(subject, **reply): ) reply = {"action": action, "reviewer": "alice", "body": "Please split the migration."} - assert answers == ([(item, reply)] if author_can_write else []) + assert answers == ([(item, reply)] if author_can_write and gate == ReviewWork.name else []) async def test_pr_open_reaches_the_work_item(druks_db): diff --git a/backend/tests/software_factory/test_webhooks_pull_request.py b/backend/tests/software_factory/test_webhooks_pull_request.py index 18df5cd74..779cbed69 100644 --- a/backend/tests/software_factory/test_webhooks_pull_request.py +++ b/backend/tests/software_factory/test_webhooks_pull_request.py @@ -85,9 +85,12 @@ async def _fresh_run(run_id): @pytest.mark.asyncio -async def test_external_merge_stores_githubs_verdict_and_ends_involvement(druks_db, tmp_path): +@pytest.mark.parametrize("gate", ["review_work", "intervene_work"]) +async def test_external_merge_stores_githubs_verdict_and_ends_involvement(druks_db, tmp_path, gate): repo, pr_number, branch = "ClawHaven/acme-app", 42, "agent/eng-1" - work_item_id, run_id = await _park_work_item(repo=repo, pr_number=pr_number, branch=branch) + work_item_id, run_id = await _park_work_item( + repo=repo, pr_number=pr_number, branch=branch, input_gate=gate + ) await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) @@ -147,12 +150,15 @@ async def test_a_redelivered_webhook_does_not_rewrite_the_verdict(druks_db, tmp_ @pytest.mark.asyncio -async def test_closed_unmerged_stores_closed_and_ends_involvement(druks_db, tmp_path): +@pytest.mark.parametrize("gate", ["review_work", "intervene_work"]) +async def test_closed_unmerged_stores_closed_and_ends_involvement(druks_db, tmp_path, gate): """A PR closed *without* merging — the operator abandoned it (e.g. deleted the branch). Store GitHub's 'closed' and un-park, so the item leaves the active board for History rather than being ignored.""" repo, pr_number, branch = "ClawHaven/acme-app", 45, "agent/eng-4" - work_item_id, run_id = await _park_work_item(repo=repo, pr_number=pr_number, branch=branch) + work_item_id, run_id = await _park_work_item( + repo=repo, pr_number=pr_number, branch=branch, input_gate=gate + ) await _fire_closed( repo=repo, diff --git a/backend/tests/test_agent_routes.py b/backend/tests/test_agent_routes.py index cffafb64e..a6ef02348 100644 --- a/backend/tests/test_agent_routes.py +++ b/backend/tests/test_agent_routes.py @@ -757,3 +757,31 @@ async def test_usage_agent_route_matches_the_service(client: TestClient, druks_d assert sum(h["runs"] for h in today["providers"]) == body["runsToday"] assert sum(h["tokens"] for h in today["providers"]) == body["tokensToday"] assert today["day"] == body["day"] + + +@pytest.mark.parametrize("control", ["repair", "recheck", "approve"]) +async def test_intervention_gate_http_controls(client: TestClient, druks_db, resume_spy, control): + note = await Note.create(body="intervention gate transport") + run = await _park(druks_db, note) + run.input_gate = "intervene_work" + run.input_request = { + "presentation": "in_app", + "label": "Implementation needs intervention", + "controls": ["repair", "recheck"], + "reply_fields": ["action", "note"], + "context": "Verification blocked", + } + await druks_db.flush() + gate = client.get(f"/api/gates/{run.id}").json() + result = client.post( + f"/api/gates/{run.id}/answer", + json={"parkedAt": gate["parkedAt"], "control": control, "note": "Service restored"}, + ) + if control == "approve": + assert result.status_code == 400 + assert not resume_spy + else: + assert result.status_code == 200 + assert resume_spy == [ + {"id": run.id, "action": control, "answers": {}, "note": "Service restored"} + ] diff --git a/docs/concepts.md b/docs/concepts.md index bbb18f2c4..7e82bd566 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -58,6 +58,30 @@ The bundled `software_factory` app owns projects, work items, ticket intake coding-agent policy, and dashboard pages. These features are examples, not platform guarantees. +### Software Factory verification and intervention + +Only a passing evaluation of the latest implementation enters work approval. +A failed evaluation uses the remaining automatic implementation rounds. A blocked +evaluation, or a failure after the round limit, parks for operator intervention. +This rule also applies when ordinary implementation approval is disabled. + +Druks keeps the PR in draft during intervention and restores draft before every +implementation rework, including changes requested by a reviewer. The operator can add an optional +note and authorize one action: + +- **Repair:** run the implementer once, then evaluate the result. +- **Recheck:** evaluate the last Druks delivery again without running the implementer. + If the PR head changed, Druks parks again. Choose Repair to deliver that changed work. + +A non-passing result parks again. An operator action does not reset the automatic +repair budget. Notes provide context; they do not override the evaluation result. +A pass returns to the configured approval and merge policy. Normal run cancellation +remains available. A GitHub review cannot answer an intervention request. + +If GitHub refuses a draft or ready-for-review transition, the run fails visibly. Druks does +not claim the PR is safely parked or continue to repair or approval. Intervention +appears as “Implementation needs intervention,” not an implementation review. + ## Durability and recovery A workflow defines either: diff --git a/frontend/src/apps/software_factory/activity.test.ts b/frontend/src/apps/software_factory/activity.test.ts index 9e56fe80a..a3a033499 100644 --- a/frontend/src/apps/software_factory/activity.test.ts +++ b/frontend/src/apps/software_factory/activity.test.ts @@ -71,3 +71,11 @@ it('does not invent old PR references or reply actions', () => { .toBe('Implementation review') expect(activity({ topic: 'pr.opened' }).label).toBe('Pull request opened') }) + + +it.each(['repair', 'recheck'])('names the intervention %s without implying approval', (action) => { + const event = { topic: 'workflow.parked', payload: { kind: 'software_factory.build', gate: 'intervene_work' } } + expect(activity(event).label).toBe('Implementation needs intervention') + const receipt = activity({ ...event, topic: 'workflow.running', payload: { ...event.payload, result: { action } } }) + expect(receipt.context).toBe(`Implementation intervention · Reply: ${action === 'repair' ? 'Repair' : 'Recheck'}`) +}) diff --git a/frontend/src/apps/software_factory/activity.ts b/frontend/src/apps/software_factory/activity.ts index 7a38e1345..891227bd1 100644 --- a/frontend/src/apps/software_factory/activity.ts +++ b/frontend/src/apps/software_factory/activity.ts @@ -20,6 +20,7 @@ function label(event: ActivityEvent): string | undefined { case 'workflow.failed': return 'Build failed' case 'workflow.cancelled': return 'Build cancelled' case 'workflow.parked': + if (event.payload?.gate === 'intervene_work') return 'Implementation needs intervention' if (event.payload?.gate === 'review_work') return 'Implementation review requested' if (event.payload?.gate === 'review') { return event.payload?.input_request?.questions?.length ? 'Clarification requested' : 'Plan review requested' @@ -43,6 +44,8 @@ const TOPIC_ICONS: Record = { } const REPLY_ACTIONS: Record = { + repair: 'Repair', + recheck: 'Recheck', approve: 'Approve', request_changes: 'Request changes', revise_contract: 'Revise contract', @@ -57,7 +60,7 @@ export function activity(event: ActivityEvent): ActivityPresentation { if (['pr.opened', 'merged', 'closed'].includes(event.topic)) { context = [facts.repo, facts.pr_number && `#${facts.pr_number}`].filter(Boolean).join(' · ') || undefined } else if (event.topic === 'workflow.running' && facts.gate) { - const gateName = facts.gate === 'review_work' ? 'Implementation review' : facts.gate === 'review' ? 'Plan review' : undefined + const gateName = facts.gate === 'intervene_work' ? 'Implementation intervention' : facts.gate === 'review_work' ? 'Implementation review' : facts.gate === 'review' ? 'Plan review' : undefined const action = facts.result?.action && REPLY_ACTIONS[facts.result.action] context = [gateName, action && `Reply: ${action}`].filter(Boolean).join(' · ') || undefined } diff --git a/frontend/src/apps/software_factory/statusLine.test.ts b/frontend/src/apps/software_factory/statusLine.test.ts index b4e2f8a02..b88d8cdf7 100644 --- a/frontend/src/apps/software_factory/statusLine.test.ts +++ b/frontend/src/apps/software_factory/statusLine.test.ts @@ -132,3 +132,10 @@ describe('runSubLine', () => { expect(runSubLine(dead, null, true)).toBe('boom') }) }) + + +it('names intervention without requesting implementation review', () => { + expect(statusLine(status({ state: 'parked', gate: 'intervene_work' }), null)).toBe( + 'Implementation needs intervention', + ) +}) diff --git a/frontend/src/apps/software_factory/statusLine.ts b/frontend/src/apps/software_factory/statusLine.ts index 8aa8f440c..682be88b7 100644 --- a/frontend/src/apps/software_factory/statusLine.ts +++ b/frontend/src/apps/software_factory/statusLine.ts @@ -9,6 +9,7 @@ import { phaseLine } from '../../lib/phase' const PARKED_LINES: Record = { review: 'Review the plan', review_work: 'Review implementation', + intervene_work: 'Implementation needs intervention', } export function parkedLine(gate: string | null): string | null { diff --git a/frontend/src/components/RunControls.test.tsx b/frontend/src/components/RunControls.test.tsx index bfeb95d34..df3121636 100644 --- a/frontend/src/components/RunControls.test.tsx +++ b/frontend/src/components/RunControls.test.tsx @@ -191,3 +191,20 @@ describe('the lent run controls', () => { ) }) }) + +it.each(['repair', 'recheck'])('submits one %s with an optional intervention note', async (control) => { + const fetchMock = stubFetch() + renderReview({ presentation: 'in_app', label: 'Implementation needs intervention', + controls: ['repair', 'recheck'], reply_fields: ['action', 'note'], + context: 'The verification service is unavailable', + }) + expect(screen.queryByRole('button', { name: 'Approve' })).toBeNull() + fireEvent.change(screen.getByPlaceholderText(/optional note/), { + target: { value: 'Service access restored' }, + }) + fireEvent.click(screen.getByRole('button', { name: control === 'repair' ? 'Repair' : 'Recheck' })) + await waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1)) + expect(JSON.parse(String(fetchMock.mock.calls[0]![1]?.body))).toEqual({ + control, answers: {}, note: 'Service access restored', + }) +}) diff --git a/frontend/src/components/RunControls.tsx b/frontend/src/components/RunControls.tsx index f400332d9..f0a896c12 100644 --- a/frontend/src/components/RunControls.tsx +++ b/frontend/src/components/RunControls.tsx @@ -81,6 +81,8 @@ export function RetryRun({ runId }: { runId: string }) { } const CONTROL_LABEL: Record = { + repair: 'Repair', + recheck: 'Recheck', approve: 'Approve', request_changes: 'Request changes', revise_contract: 'Revise contract',