diff --git a/backend/druks/agents.py b/backend/druks/agents.py index 18ee7e4..2ef0a80 100644 --- a/backend/druks/agents.py +++ b/backend/druks/agents.py @@ -156,7 +156,7 @@ async def _do() -> Any: # a standalone run is its own memoized step + session ) # Body-level only, on both passes: in-step calls hit the early return # live and are skipped with their step on replay. - workflow.record._append(result) + workflow.journal.add(result) return result async def _run(self, *, workflow_id: str, **context: Any) -> Any: diff --git a/backend/druks/build/contracts.py b/backend/druks/build/contracts.py index 173f584..bec82b3 100644 --- a/backend/druks/build/contracts.py +++ b/backend/druks/build/contracts.py @@ -1,4 +1,4 @@ -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal, cast from pydantic import BaseModel, Field, model_validator @@ -8,16 +8,24 @@ HumanFeedbackAction, ReviewDecision, ) +from druks.workflows import Gate, Workflow +if TYPE_CHECKING: + from druks.build.workflows import BuildWorkflow -class HumanFeedback(BaseModel): - reviewer: str - body: str = "" - status: Literal["pending", "triaged"] = "pending" - triage_action: HumanFeedbackAction | None = None - triage_body: str = "" - question: str = "" - implementation_instructions: str = "" + +# The PR webhook resumes approve/request_changes; revise_contract and cancel +# come from the operator's UI. +class ReviewWork(Gate): + action: Literal["approve", "request_changes", "revise_contract", "cancel"] + reviewer: str | None = None + body: str | None = None + + @classmethod + async def on_wait(cls, workflow: Workflow) -> None: + build = cast("BuildWorkflow", workflow) + await build._set_pr_draft(draft=False) + await build._request_assignee_review() class RepoProfilerOutput(AgentOutput): @@ -77,6 +85,8 @@ class PlanData(BaseModel): plan_markdown: str = "" questions: list[QuestionOutput] = Field(default_factory=list) acceptance_criteria: list[AcceptanceCriterionOutput] = Field(default_factory=list) + # Resolved by the planner; a revision carries None. + assignee_github_login: str | None = None def get_answered(self, picks: dict[str, str]) -> list[dict[str, str]]: # Each question the operator answered, paired with its answer — what the @@ -99,6 +109,9 @@ class PlanOutput(AgentOutput): plan_markdown: str acceptance_criteria: list[AcceptanceCriterionOutput] questions: list[QuestionOutput] = Field(max_length=8) + # Required but nullable: the planner always reports the field, null when it + # resolved no assignee login convincingly. + assignee_github_login: str | None def get_artifact(self) -> dict[str, str]: return {"kind": "markdown", "title": "Implementation plan", "content": self.plan_markdown} @@ -108,6 +121,7 @@ def to_result(self) -> PlanData: plan_markdown=self.plan_markdown, acceptance_criteria=self.acceptance_criteria, questions=self.questions, + assignee_github_login=self.assignee_github_login, ) @@ -130,17 +144,10 @@ def to_result(self) -> PlanData: class ReviewOutput(AgentOutput): - # The plan-review agent can't COMMENT — that domain value is for human PR - # reviews, so the contract lists only the three the agent may return. - decision: Literal[ - ReviewDecision.APPROVE, - ReviewDecision.APPROVE_WITH_REQUIRED_CHANGES, - ReviewDecision.REQUEST_CHANGES, - ] + # No get_artifact: the plan must stay the parked ask's resolved document; + # the fallback park sends the critique as ask context instead. + decision: Literal[ReviewDecision.APPROVE, ReviewDecision.REQUEST_CHANGES] body: str - # Required but nullable: the agent always reports the field, null when it - # resolved no assignee login convincingly. - assignee_github_login: str | None class TriageOutput(AgentOutput): diff --git a/backend/druks/build/enums.py b/backend/druks/build/enums.py index e57a163..38abb73 100644 --- a/backend/druks/build/enums.py +++ b/backend/druks/build/enums.py @@ -3,7 +3,6 @@ class ReviewDecision(StrEnum): APPROVE = "APPROVE" - APPROVE_WITH_REQUIRED_CHANGES = "APPROVE_WITH_REQUIRED_CHANGES" REQUEST_CHANGES = "REQUEST_CHANGES" COMMENT = "COMMENT" diff --git a/backend/druks/build/journal.py b/backend/druks/build/journal.py index ab76859..234ce20 100644 --- a/backend/druks/build/journal.py +++ b/backend/druks/build/journal.py @@ -1,83 +1,60 @@ -from dataclasses import dataclass, field +from contextlib import suppress from druks.build.contracts import ( EvaluationOutput, - HumanFeedback, ImplementationOutput, PlanData, - ReviewOutput, + ReviewWork, + TriageOutput, ) -from druks.build.enums import ReviewDecision +from druks.workflows import Journal -@dataclass -class PlanRecord: - plan: PlanData - reviews: list[ReviewOutput] = field(default_factory=list) - - -class BuildJournal: - """The run's working memory, rebuilt on replay. Mutate only from workflow-body - code after a memoized call — never inside a @step, where replay skips the - write.""" - - def __init__(self) -> None: - self.plans: list[PlanRecord] = [] - self.implementations: list[ImplementationOutput] = [] - self.evaluations: list[EvaluationOutput] = [] - self.human_feedback: list[HumanFeedback] = [] - - def add_plan(self, plan: PlanData) -> PlanData: - self.plans.append(PlanRecord(plan=plan)) - return plan - - def add_plan_review(self, review: ReviewOutput) -> ReviewOutput: - self.plans[-1].reviews.append(review) - return review - - def add_implementation(self, implementation: ImplementationOutput) -> ImplementationOutput: - self.implementations.append(implementation) - return implementation - - def add_evaluation(self, evaluation: EvaluationOutput) -> EvaluationOutput: - self.evaluations.append(evaluation) - return evaluation - - def add_feedback(self, feedback: HumanFeedback) -> HumanFeedback: - self.human_feedback.append(feedback) - return feedback - +class BuildJournal(Journal): @property def plan(self) -> PlanData: - return self.plans[-1].plan if self.plans else PlanData() + return self.latest(PlanData) or PlanData() @property - def last_implementation(self) -> ImplementationOutput | None: - return self.implementations[-1] if self.implementations else None + def plan_revision(self) -> int: + return len(self.filter(PlanData)) @property - def plan_revision(self) -> int: - return len(self.plans) + def implementations(self) -> list[ImplementationOutput]: + return self.filter(ImplementationOutput, status="success") + + @property + def last_implementation(self) -> ImplementationOutput | None: + with suppress(IndexError): + return self.implementations[-1] + return @property def implementation_revision(self) -> int: return len(self.implementations) + @property + def evaluations(self) -> list[EvaluationOutput]: + return self.filter(EvaluationOutput) + @property def assignee_github_login(self) -> str | None: - for record in reversed(self.plans): - for review in reversed(record.reviews): - if review.assignee_github_login: - return review.assignee_github_login - return None + for plan in reversed(self.filter(PlanData)): + if plan.assignee_github_login: + return plan.assignee_github_login + return - def reviewer_requirements(self) -> list[ReviewOutput]: - # Approve-with-required-changes verdicts on the current plan draft only — - # reviews of superseded drafts don't bind the implementer. - current = self.plans[-1].reviews if self.plans else [] + @property + def human_feedback(self) -> list[dict[str, str]]: + # zip strict=False: the newest reply is still undigested while its own + # triage agent renders this very projection. + replies = self.filter(ReviewWork, action="request_changes") return [ - review - for review in current - if review.decision == ReviewDecision.APPROVE_WITH_REQUIRED_CHANGES - and review.body.strip() + { + "reviewer": reply.reviewer or "(triage)", + "body": triage.body, + "question": triage.question, + "implementation_instructions": triage.implementation_instructions, + } + for reply, triage in zip(replies, self.filter(TriageOutput), strict=False) ] diff --git a/backend/druks/build/workflows.py b/backend/druks/build/workflows.py index 09bac95..bff526f 100644 --- a/backend/druks/build/workflows.py +++ b/backend/druks/build/workflows.py @@ -1,11 +1,11 @@ import logging from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, Literal, cast +from typing import TYPE_CHECKING, Any from pydantic import BaseModel, Field from druks.accounts.models import Account -from druks.build.contracts import HumanFeedback, ImplementationOutput +from druks.build.contracts import ImplementationOutput, ReviewWork from druks.build.enums import ( EvaluationVerdict, HumanFeedbackAction, @@ -24,7 +24,7 @@ from druks.settings import load_settings from druks.skills.models import Skill from druks.ticketing.enums import SemanticStatus -from druks.workflows import FatalError, Gate, Workflow, step +from druks.workflows import FatalError, Workflow, step from .extension import Build from .journal import BuildJournal @@ -45,22 +45,6 @@ GITHUB_MCP_URL = "https://api.githubcopilot.com/mcp/" -# Build's one external gate: code review happens on the PR. `action` spans the webhook's -# review vocab (approve, request_changes — the only actions a resumer can send) plus the -# operator's UI decisions (revise_contract, cancel). on_wait un-drafts the PR and pings -# the assignee. -class ReviewWork(Gate): - action: Literal["approve", "request_changes", "revise_contract", "cancel"] - reviewer: str | None = None - body: str | None = None - - @classmethod - async def on_wait(cls, workflow: Workflow) -> None: - build = cast("BuildWorkflow", workflow) - await build._set_pr_draft(draft=False) - await build._request_assignee_review() - - @dataclass(frozen=True, kw_only=True) class BuildWorkspace(RepoWorkspace): # The base RepoWorkspace brings the cloned repo + token; a build run adds the @@ -91,6 +75,8 @@ def get_agent_run_kwargs(self, **kwargs: Any) -> dict[str, Any]: class BuildWorkflow(Workflow): steps_reuse_sandbox = True workspace_class = BuildWorkspace + journal_class = BuildJournal + journal: BuildJournal class Settings(BaseModel): auto_dispatch_on_plan_approval: bool = Field( @@ -111,10 +97,6 @@ class Settings(BaseModel): description="Run the line-level reviewer after a passing evaluation.", ) - def __init__(self) -> None: - super().__init__() - self._journal = BuildJournal() - @classmethod async def dispatch( cls, @@ -234,7 +216,7 @@ async def get_prompt_context(self, **context: Any) -> dict[str, Any]: task_owner_name=self.input.task_owner_name, task_owner_email=self.input.task_owner_email, related_repos=self._related_repos(), - journal=self._journal, + journal=self.journal, ) return { "verification": await self._policy.verification_block( @@ -261,24 +243,31 @@ async def _load_settings(self) -> "BuildWorkflow.Settings": return self.settings() async def _plan_phase(self) -> bool: - """Plan → questions? park for the operator's answers and re-plan : approve. - True → implement; cancel raises.""" + """Gate mode: plan → park. Auto mode: machine review with one bounded + redraft. True → implement; cancel raises.""" + gate = self._policy.plan_approval_gate(self._settings.auto_dispatch_on_plan_approval) answered: list[dict[str, str]] = [] note = "" while True: - plan = self._journal.add_plan( - await Build.generate_plan(answered_questions=answered, operator_note=note) - ) - # No open questions and a clean grade under an auto-approve policy ships - # the plan without asking the operator. - if not plan.questions: - grade = self._journal.add_plan_review(await Build.review_plan()) - if grade.decision == ReviewDecision.APPROVE and ( - self._policy.plan_approval_gate(self._settings.auto_dispatch_on_plan_approval) - == "none" - ): + reviewer_notes = "" + redrafted = False + critique = "" + while True: + plan = await Build.generate_plan( + answered_questions=answered, operator_note=note, reviewer_notes=reviewer_notes + ) + if gate != "none" or plan.questions: + break + grade = await Build.review_plan() + if grade.decision == ReviewDecision.APPROVE: return True - reply = await self.review(questions=plan.questions) + if redrafted: + # Exhausted — park below, critique on the ask. + critique = grade.body + break + redrafted = True + reviewer_notes = grade.body + reply = await self.review(questions=plan.questions, context=critique) if reply.action == "cancel": raise FatalError("cancelled at plan review") if reply.action == "approve" and not plan.questions: @@ -289,7 +278,7 @@ async def _plan_phase(self) -> bool: async def _implement_phase(self) -> None: while True: await self.implement() - evaluation = self._journal.add_evaluation(await Build.evaluate_implementation()) + evaluation = await Build.evaluate_implementation() if evaluation.verdict == EvaluationVerdict.PASS: if self._settings.review_code: await Build.review_code() @@ -297,7 +286,7 @@ async def _implement_phase(self) -> None: return continue if evaluation.verdict == EvaluationVerdict.FAIL and ( - self._journal.implementation_revision < self._settings.max_implementation_revisions + self.journal.implementation_revision < self._settings.max_implementation_revisions ): continue if await self._work_gate(): @@ -315,9 +304,9 @@ async def _work_gate(self) -> bool: if decision.action == "approve": return await self._approved_work() if decision.action == "request_changes": - return await self._triage(decision) + return await self._triage() if decision.action == "revise_contract": - self._journal.add_plan(await Build.revise_contract()) + await Build.revise_contract() return False if decision.action == "cancel": await self._push_ticket_status(SemanticStatus.CANCELED) @@ -332,20 +321,20 @@ async def _approved_work(self) -> bool: await self._clear_draft() return True - async def _triage(self, decision: ReviewWork) -> bool: - feedback = self._journal.add_feedback(await self.triage_feedback(decision)) - if feedback.triage_action == HumanFeedbackAction.CHANGE_REQUIRED: + async def _triage(self) -> bool: + feedback = await Build.triage_human_feedback() + if feedback.action == HumanFeedbackAction.CHANGE_REQUIRED: return False # loop → implement - if feedback.triage_action == HumanFeedbackAction.CONTRACT_CHANGE_REQUIRED: - self._journal.add_plan(await Build.revise_contract()) + if feedback.action == HumanFeedbackAction.CONTRACT_CHANGE_REQUIRED: + await Build.revise_contract() return False - if feedback.triage_action == HumanFeedbackAction.CLOSE: + if feedback.action == HumanFeedbackAction.CLOSE: raise FatalError("closed at human triage") # NO_CHANGE / QUESTION → re-park return await self._work_gate() - # Body code, never @step: the agent calls inside memoize themselves, and the - # journal writes + set_state must re-run on replay — a @step would skip them. + # Body code, never @step: the agent calls inside memoize themselves and land + # on the record, and set_state must re-run on replay — a @step would skip them. async def implement(self) -> ImplementationOutput: delivery = await Build.implement() # A bail is a stop, not a result: the implementer hit a contradiction in the @@ -357,18 +346,7 @@ async def implement(self) -> ImplementationOutput: # First delivery: the implementer provisioned the branch + draft PR alongside # its commits; publish the pair (the run.state signal mirrors it onto the item). await self.set_state(branch=delivery.branch, pr_number=delivery.pr_number) - return self._journal.add_implementation(delivery) - - async def triage_feedback(self, decision: ReviewWork) -> HumanFeedback: - parsed = await Build.triage_human_feedback() - return HumanFeedback( - reviewer=decision.reviewer or "(triage)", - body=parsed.body, - triage_action=parsed.action, - triage_body=parsed.body, - question=parsed.question, - implementation_instructions=parsed.implementation_instructions, - ) + return delivery @step async def merge(self) -> None: @@ -424,7 +402,7 @@ async def _push_ticket_status(self, status: SemanticStatus) -> None: await work_item.set_remote_status(status) async def _request_assignee_review(self) -> None: - login = self._journal.assignee_github_login + login = self.journal.assignee_github_login if login and self.input.repo and self.pr_number: try: await get_github_client(load_settings()).request_pull_request_reviewers( diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 6bc96bd..06eddef 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -50,9 +50,9 @@ "AgentCallStatus", "FatalError", "Gate", - "ReviewReply", + "Journal", + "OperatorReply", "Run", - "RunRecord", "RunState", "SubjectActivity", "SubjectSummary", @@ -165,18 +165,18 @@ def _kind_from_class_name(name: str) -> str: return _CAMEL_BOUNDARY.sub("_", name).lower() -class RunRecord: - """The run's typed history — every body-level agent output and gate reply, - appended by the platform in call order, rebuilt identically on replay. - Authors project working state off it with ``list``/``latest``.""" +class Journal: + """The run's working memory. Druks adds each body-level agent output and + gate reply; a body adds its own derived values; subclass it for named + projections and declare yours via ``journal_class``.""" def __init__(self) -> None: self._entries: list[Any] = [] - def _append(self, entry: Any) -> None: + def add(self, entry: Any) -> None: self._entries.append(entry) - def list(self, contract: type[T], **filters: Any) -> list[T]: + def filter(self, contract: type[T], **filters: Any) -> list[T]: return [ entry for entry in self._entries @@ -186,7 +186,7 @@ def list(self, contract: type[T], **filters: Any) -> list[T]: def latest(self, contract: type[T], **filters: Any) -> T | None: with suppress(IndexError): - return self.list(contract, **filters)[-1] + return self.filter(contract, **filters)[-1] return @@ -235,11 +235,11 @@ async def _on_wait() -> None: await DBOS.run_step_async(StepOptions(name=f"{cls.topic}._on_wait"), _on_wait) payload = await _park(workflow, cls.topic, input_request, ttl_seconds) reply = cls.model_validate(payload) - workflow.record._append(reply) + workflow.journal.add(reply) return reply -class ReviewReply(Gate): +class OperatorReply(Gate): """The operator's in-app review decision. Answers and note are content for the next agent prompt, never control flow.""" @@ -454,6 +454,8 @@ class Workflow: steps_reuse_sandbox: ClassVar[bool] = False # The Workspace subclass agents run in; an extension sets it (default: the bare VM). workspace_class: ClassVar[type[Workspace]] = Workspace + # The Journal subclass the run keeps; an extension sets it for named projections. + journal_class: ClassVar[type[Journal]] = Journal # Exactly one: run() is a single operation, auto-stepped, no ceremony. # run_multistep() orchestrates explicit @step calls and/or gates. run: ClassVar[Callable] @@ -517,9 +519,7 @@ def __init__(self) -> None: # Who requested/triggered the run, replayed off the reserved input # key; None on system-owned runs (crons, old checkpoints). self.account_id: str | None = None - # The run's typed history, appended at the platform chokepoints — - # read, don't write; project with list/latest. - self.record = RunRecord() + self.journal = self.journal_class() # Facts published with set_state, kept warm for sync reads (templates, # workspace kwargs); the durable copy is the run's DBOS events. self._state_facts: dict[str, Any] = {} @@ -556,9 +556,12 @@ async def _fan_out() -> None: await DBOS.run_step_async(StepOptions(name="run.state", **_IO_RETRIES), _fan_out) - async def review(self, *, questions: list[BaseModel] | None = None) -> ReviewReply: - # Park for an in-app decision: the ask carries the controls and any - # questions; the ReviewReply carries the operator's answer (the resume + async def review( + self, *, questions: list[BaseModel] | None = None, context: str = "" + ) -> OperatorReply: + # Park for an in-app decision: the ask carries the controls, any + # questions, and optional context (rendered beside the reviewed + # document); the OperatorReply carries the operator's answer (the resume # endpoint holds the action to the offered controls). External gates use # a Gate. The artifact the reviewer judges isn't named here — a parked # run can't produce new ones, so the read side resolves the latest on @@ -566,15 +569,17 @@ async def review(self, *, questions: list[BaseModel] | None = None) -> ReviewRep if not self.subject: # An in-app review is answered from the subject's surfaces; a # subjectless run has none, so nobody would ever see the ask. - raise SubjectlessGate(ReviewReply.topic) + raise SubjectlessGate(OperatorReply.topic) request = { "presentation": "in_app", "controls": list(_REVIEW_CONTROLS), "questions": [q.model_dump(mode="json") for q in questions or ()], } - payload = await _park(self, ReviewReply.topic, request, GATE_TTL_SECONDS) - reply = ReviewReply.model_validate(payload) - self.record._append(reply) + if context: + request["context"] = context + payload = await _park(self, OperatorReply.topic, request, GATE_TTL_SECONDS) + reply = OperatorReply.model_validate(payload) + self.journal.add(reply) return reply async def get_prompt_context(self, **context: Any) -> dict[str, Any]: diff --git a/backend/templates/prompts/build/build_workflow/_header.md b/backend/templates/prompts/build/build_workflow/_header.md index e06170a..1a2c107 100644 --- a/backend/templates/prompts/build/build_workflow/_header.md +++ b/backend/templates/prompts/build/build_workflow/_header.md @@ -44,18 +44,6 @@ **Verification:** {{ ac.verification.strip() }} {% endif %} -{% endfor %} -{% endif %} -{% if build.journal.reviewer_requirements() %} -## Reviewer requirements (active) - -These are binding plan clarifications from the plan reviewer. Treat them as part of the plan and apply them. - -{% for req in build.journal.reviewer_requirements() %} -### Requirement {{ loop.index }} - -{{ req.body.strip() }} - {% endfor %} {% endif %} {% if build.journal.evaluations %} @@ -74,7 +62,7 @@ These are binding plan clarifications from the plan reviewer. Treat them as part ## Human feedback {% for fb in build.journal.human_feedback %} -### {{ fb.reviewer }} — status: {{ fb.status }} +### {{ fb.reviewer }} {% if fb.body %} **Body:** {{ fb.body.strip() }} diff --git a/backend/templates/prompts/build/build_workflow/evaluate_implementation.md b/backend/templates/prompts/build/build_workflow/evaluate_implementation.md index 31da2c2..24412c7 100644 --- a/backend/templates/prompts/build/build_workflow/evaluate_implementation.md +++ b/backend/templates/prompts/build/build_workflow/evaluate_implementation.md @@ -1,7 +1,7 @@ # Contract Verifier You are the contract verifier making a factual determination: does this diff satisfy each -acceptance criterion and each binding reviewer requirement? You have no opinion about what the +acceptance criterion of the reviewed plan? You have no opinion about what the plan should have asked for. You verify what it did ask for, exhaustively, in a single pass. ## Core truths @@ -11,15 +11,16 @@ plan should have asked for. You verify what it did ask for, exhaustively, in a s better solutions to the problem. - **Exhaustiveness is your primary obligation.** A finding you omit now costs a full implementation round to surface next round. The system will not return for a second pass on - the same diff. Walk all seven sweeps before returning. -- **You cannot invent new requirements.** Something concerning in the diff that no AC or - reviewer requirement covers belongs under "Follow-up recommendations," not in blocking + the same diff. Walk all six sweeps before returning. +- **You cannot invent new requirements.** Something concerning in the diff that no AC + covers belongs under "Follow-up recommendations," not in blocking findings. The only exceptions: a regression the implementer just introduced in this specific revision (quote the new code that caused it), or a clear security flaw with a data-loss or privilege-escalation path. -- **The plan was reviewed and approved.** You do not re-evaluate whether the approach was - right, whether the tech choices were ideal, or whether you would have specified it - differently. You verify whether it was implemented as specified. +- **The plan arrives complete.** It was reviewed before implementation started — any reviewer + critique was already folded into the plan you are reading. You do not re-evaluate whether + the approach was right, whether the tech choices were ideal, or whether you would have + specified it differently. You verify whether it was implemented as specified. ## Boundaries @@ -32,17 +33,16 @@ plan should have asked for. You verify what it did ask for, exhaustively, in a s {% include "build/build_workflow/_header.md" %} {% include "build/build_workflow/_related_repos.md" %} {% include "build/build_workflow/_skills.md" %} -Evaluate the implementation against the **Current plan** above, entries in the **Reviewer requirements (active)** section above (binding plan clarifications), the issue, and the current PR diff. Treat reviewer requirements as part of plan compliance: a missed reviewer requirement is a fail just like a missed acceptance criterion. When `base_sha` and `head_sha` are listed in the **Workflow context** section above, use them as the authoritative diff range and evaluate `head_sha` against `base_sha`. 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. Evaluate every acceptance criterion from the PR state and report one result per criterion. Inspection commands such as git diff/show, rg, and sed are allowed for review. Verification commands are different: run only the configured verification profile commands when feasible and report their results in checks. Do not invent repo-specific smoke tests or package install commands. Return exactly one final result object. Return pass only when the work is ready for a human final PR review. Return fail for actionable implementation changes. +Evaluate the implementation against the **Current plan** above, the issue, and the current PR diff. When `base_sha` and `head_sha` are listed in the **Workflow context** section above, use them as the authoritative diff range and evaluate `head_sha` against `base_sha`. 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. Evaluate every acceptance criterion from the PR state and report one result per criterion. Inspection commands such as git diff/show, rg, and sed are allowed for review. Verification commands are different: run only the configured verification profile commands when feasible and report their results in checks. Do not invent repo-specific smoke tests or package install commands. Return exactly one final result object. Return pass only when the work is ready for a human final PR review. Return fail for actionable implementation changes. EXHAUSTIVE ENUMERATION — this is the single most important rule. Subsequent rounds will not retry, and findings you omit now cost an entire extra implementation loop to surface next round. Walk through these sweeps and list every blocker you find in a single response: 1. Each acceptance criterion explicitly — does the diff satisfy it? -2. Each active reviewer_requirement explicitly — does the diff satisfy it? +2. Any preference or implementation approach the plan explicitly named (e.g. parser-based vs regex-based, immutability, allowlist scope) — even if the current implementation works, it must match the approach the plan asked for. 3. Tests covering every changed code path — gap = blocker. 4. Dependency / lockfile changes — pinning, format, version compatibility. 5. Input validation + error handling boundaries the change introduces. -6. Any preference or implementation approach the reviewer_requirements explicitly named (e.g. parser-based vs regex-based, immutability, allowlist scope) — even if the current implementation works, it must match the approach reviewers asked for. -7. Side effects: imports, generated files, lockfiles, config changes outside the stated scope. -Do not return until you have collected every finding you can identify across all seven sweeps. The implementer fixes verbatim from your findings list, so anything missing here forces another full revision round. +6. Side effects: imports, generated files, lockfiles, config changes outside the stated scope. +Do not return until you have collected every finding you can identify across all six sweeps. The implementer fixes verbatim from your findings list, so anything missing here forces another full revision round. UNFULFILLABLE-AC GATE — before scoring any finding against an acceptance criterion, check whether the criterion is **code-verifiable** by you (reading the diff, inspecting tests, running the configured verification profile). If a criterion requires manual operator action — "manually smoke X", "load the app locally", "verify visually in the browser", "click through Y", "confirm against the live N integration", "screenshot the rendered output", etc. — it is **not satisfiable by the implementer** through any code change. Mark its `acceptance_results` entry as `not_run` with a one-line reason ("requires operator-driven manual smoke") and do NOT emit a blocking finding against it. The planner is supposed to keep these out of binding AC, but if one slips through, the evaluator must not loop the implementer over it forever. Report once per round at most, as a `low`-severity note recommending the operator smoke post-merge — never as `high` or `medium`. @@ -53,10 +53,10 @@ INFEASIBLE-BLOCKER GATE — return `blocked`, NOT `fail`, when the only thing ke The test is strict and binary: *can a code change the implementer is allowed to make resolve this blocker?* Yes → `fail` with an actionable finding (a test its own diff broke, a missed AC, a real in-scope code defect). No → `blocked`. Never loop the implementer on a blocker no code change can clear. SEVERITY CALIBRATION — assign severity per finding: -- high: correctness bug, security flaw, data loss, crash, or a directly missed acceptance criterion / reviewer requirement. +- high: correctness bug, security flaw, data loss, crash, or a directly missed acceptance criterion. - medium: missing test coverage for an AC, contract violation that won't crash but weakens guarantees, lockfile/dependency hygiene that affects reproducibility. - low: style preference, naming, formatting, refactor suggestion where the current implementation is correct and meets all stated requirements. A finding only qualifies as low if shipping the PR as-is would not break the contract — Druks will surface low findings as review notes on the merged PR rather than burning an implementation loop on them. -When in doubt between low and medium, prefer medium. Mark anything that maps to an AC or reviewer_requirement as medium or high — never low. Findings that are all low severity are never a fail verdict: return pass and let them ride as review notes. +When in doubt between low and medium, prefer medium. Mark anything that maps to an AC as medium or high — never low. Findings that are all low severity are never a fail verdict: return pass and let them ride as review notes. SUBSTANTIAL PROGRESS — when you flagged a finding in a prior round AND the implementer's revision substantively addresses the spirit of that requirement, that finding is resolved, even if you can identify a subtler edge case within the same theme. Subtler edges of an already-substantively-fixed theme become PR-review notes for the human reviewer (mention them in the body), NOT blocking findings that loop the implementer. Demanding perfection on a theme that has been substantially addressed costs an entire revision round for marginal value — ship-then-followup is cheaper. Concrete examples of the trap: 'body isolation is mostly done but body text might appear in Python traceback locals', 'safety_flags are validated but list ordering is platform-dependent', 'logging discipline is honored everywhere except one debug line that is gated behind DEBUG=true'. These are notes, not blockers. A new blocker across rounds must be on a DIFFERENT theme or be a freshly-introduced correctness/security bug. diff --git a/backend/templates/prompts/build/build_workflow/generate_plan.md b/backend/templates/prompts/build/build_workflow/generate_plan.md index 215541d..7381edc 100644 --- a/backend/templates/prompts/build/build_workflow/generate_plan.md +++ b/backend/templates/prompts/build/build_workflow/generate_plan.md @@ -50,6 +50,14 @@ The operator requested changes on your previous plan in their own words. The blo > {{ operator_note | replace("\n", "\n> ") }} +{% endif %} +{% if reviewer_notes %} +## Plan reviewer critique + +The plan reviewer rejected your previous draft with the critique below. Fold every point into this redraft — the reviewer never edits the plan; you produce the complete corrected plan yourself. If the redraft still fails review, the run parks for the operator — resolve every point now: + +> {{ reviewer_notes | replace("\n", "\n> ") }} + {% endif %} Generate the initial implementation plan. Include open questions only when the plan cannot be made decision-complete from the issue and repository build. Return specific acceptance criteria describing what must be true for this PR to pass. When the work changes a protocol or wire contract, include exact request/response examples in the plan or acceptance criteria. If the verification profile is empty, do not add standalone test/lint/typecheck criteria; keep verification criteria tied to commands that are actually configured or explicitly requested by the issue. @@ -63,3 +71,11 @@ When the source ticket asks for a manual smoke or visual check, do ONE of these Smoke / manual-verify requests in the source are operator concerns, not implementer concerns. Honor the intent (the operator wants to test the UX) without making the agent loop block on something it can't satisfy. When you fetch the ticket, its description ends with a `# Druks scope brief` heading — that section is the authoritative scope summary (problem, scope, acceptance criteria, out of scope). Everything above that heading is the human-authored source material; use it as detail and context, but the brief section wins on intent and shape. If the source lists per-test acceptance criteria that the brief summarises into prose, the source is canonical for those tests — do not drop them. + +ASSIGNEE RESOLUTION — the `assignee_github_login` schema field. Resolve the ticket +assignee's GitHub login via the github MCP from their name +`{{ build.task_owner_name or "(unknown)" }}` or email +`{{ build.task_owner_email or "(unknown)" }}` (user search; pick the +candidate whose profile clearly matches). Report the login string, or `null` when +nothing resolves convincingly — never guess. Druks uses it to request their +review at the parks that await a human; do not request reviewers yourself. diff --git a/backend/templates/prompts/build/build_workflow/implement.md b/backend/templates/prompts/build/build_workflow/implement.md index bbb981a..6e28c77 100644 --- a/backend/templates/prompts/build/build_workflow/implement.md +++ b/backend/templates/prompts/build/build_workflow/implement.md @@ -33,7 +33,7 @@ acceptance criterion — nothing more, nothing less. {% include "build/build_workflow/_header.md" %} {% include "build/build_workflow/_related_repos.md" %} {% include "build/build_workflow/_skills.md" %} -Implement the approved plan (rendered above as **Current plan**) on the work branch (see "Delivering your work" below). Entries in the **Reviewer requirements (active)** section above are binding plan clarifications from the plan reviewer; treat them as part of the plan and apply them. If the **Human feedback** section above carries a triaged entry with implementation instructions, apply those instructions as the current revision request. Do not run ad hoc install, lint, test, build, or smoke commands during implementation unless the plan explicitly requires changing those commands or generated outputs; verification is handled by the evaluator. Never leave dependency lockfile, generated, or cache changes unless they are part of the requested implementation. Return structured evidence for every acceptance criterion, every check you ran or intentionally did not run, changed files, and known risks. If a check was not run, include status not_run and a reason. +Implement the approved plan (rendered above as **Current plan**) on the work branch (see "Delivering your work" below). If the **Human feedback** section above carries an entry with implementation instructions, apply those instructions as the current revision request. Do not run ad hoc install, lint, test, build, or smoke commands during implementation unless the plan explicitly requires changing those commands or generated outputs; verification is handled by the evaluator. Never leave dependency lockfile, generated, or cache changes unless they are part of the requested implementation. Return structured evidence for every acceptance criterion, every check you ran or intentionally did not run, changed files, and known risks. If a check was not run, include status not_run and a reason. ## Delivering your work diff --git a/backend/templates/prompts/build/build_workflow/review_code.md b/backend/templates/prompts/build/build_workflow/review_code.md index 7e08dc2..c5f23a2 100644 --- a/backend/templates/prompts/build/build_workflow/review_code.md +++ b/backend/templates/prompts/build/build_workflow/review_code.md @@ -45,7 +45,7 @@ WHAT TO LOOK FOR — beyond AC correctness, which is the evaluator's job, not yo - Comments and naming that lie or mislead — drift between what the comment claims and what the code does. WHAT NOT TO FLAG: -- Anything the evaluator already covered (AC correctness, reviewer requirements). Don't relitigate. +- Anything the evaluator already covered (AC correctness). Don't relitigate. - Style nits that don't change behavior or maintainability (whitespace, alphabetical import order, single vs double quote unless the repo enforces one). - "I would have done this differently" without a concrete reason tied to maintainability, performance, or correctness. - Pre-existing issues in unchanged code. You're reviewing the diff, not the codebase. diff --git a/backend/templates/prompts/build/build_workflow/review_plan.md b/backend/templates/prompts/build/build_workflow/review_plan.md index 03c7b32..cc4affe 100644 --- a/backend/templates/prompts/build/build_workflow/review_plan.md +++ b/backend/templates/prompts/build/build_workflow/review_plan.md @@ -2,34 +2,33 @@ You are the principal engineer doing the gate review before implementation starts. The planner thinks this plan is ready. Your job is to catch the shape problems that are cheap to fix now -and catastrophically expensive after implementation starts. +and catastrophically expensive after implementation starts. No human will see this plan before +implementation — your verdict is the gate. ## Core truths - **Shape before details.** Wrong architecture, wrong scope, wrong layer — these make every downstream step more expensive. If the shape is wrong, REQUEST_CHANGES immediately rather than polishing the details. -- **You own the cost of your requirements.** Every binding requirement you write will be handed - verbatim to the implementer and audited verbatim by the evaluator for every revision round. - If a requirement names a specific API call that doesn't exist in this version of the - framework, the implementer either blindly breaks something or burns rounds discovering the - dead end. The evaluator then enforces your wrong requirement for as long as it remains in - the contract. -- **Verify before prescribing.** Before naming an exact method call or library API in a - binding requirement, grep the repo for prior usage or read adjacent code that does similar - work. If you cannot verify it exists in this codebase, write the behavior instead: "read the +- **You own the cost of your critique.** On REQUEST_CHANGES the planner redrafts the plan once, + folding your critique verbatim; the implementer and evaluator then work from that redrafted + plan. If your critique names a specific API call that doesn't exist in this version of the + framework, the redraft bakes the mistake into the contract for every round that follows. +- **Verify before prescribing.** Before naming an exact method call or library API in your + critique, grep the repo for prior usage or read adjacent code that does similar work. If you + cannot verify it exists in this codebase, write the behavior instead: "read the request body asynchronously" not "use `await request.abody()`". -- **Your requirements are the contract.** The plan will NOT be regenerated after your review. - What you write appends verbatim as binding constraints. Every requirement must be concrete, - self-contained, and achievable in this codebase. +- **One shot, then a human.** If the redrafted plan still doesn't pass your re-review, the run + parks for the operator with your critique attached. Make every point concrete and + self-contained — both the planner and, on the fallback, a human will act on it directly. ## Boundaries -- You are not the implementer. Specify what the code must achieve, not line-by-line how to +- You are not the implementer. Specify what the plan must achieve, not line-by-line how to write it. -- REQUEST_CHANGES triggers an expensive full re-plan. Use it when the architecture is wrong, - not when details need clarification. Use APPROVE_WITH_REQUIRED_CHANGES for the latter. -- Do not add requirements beyond what the issue and the existing codebase support. +- You are not the planner either: never rewrite the plan yourself — the critique is your whole + output, and the planner folds it. +- Do not require changes beyond what the issue and the existing codebase support. {% include "build/build_workflow/_header.md" %} {% include "build/build_workflow/_related_repos.md" %} @@ -43,23 +42,14 @@ SCOPE & APPROACH REVIEW — do this BEFORE evaluating contract details. These ar - Implied follow-ups the plan didn't mention: does this change require docs, a migration, an admin/CLI affordance, a feature flag, or a backfill that the plan silently omitted? Either fold them in or call them out as explicit out-of-scope so the operator can decide. Pick exactly one decision: -- APPROVE: the plan is decision-complete, correctly scoped, approach matches repo patterns, and is implementable with no required edits. -- APPROVE_WITH_REQUIRED_CHANGES: the overall plan direction is sound but the implementation contract needs binding clarifications (exact wire schemas, parser boundaries, error code taxonomy, daemon/gateway contracts, side-effect boundaries, exact command.result and command.error shapes, malformed-payload behavior, etc.). Also use this when the scope is right but the approach needs a concrete nudge toward an existing repo pattern that won't re-shape the plan. The plan will NOT be re-generated; your review body is appended verbatim as binding Reviewer Requirements that the implementer and evaluator must follow. Make every requirement concrete and self-contained — do not say 'clarify X', say exactly what X must be. -- REQUEST_CHANGES: reserved for major plan flaws only — wrong architecture, scope that should be split into multiple PRs, approach that fights the codebase's existing patterns, missing/incorrect acceptance criteria, unsafe scope, wrong auth/security boundary, or an unimplementable plan. This triggers an expensive full re-plan, so use it sparingly — but do use it when the shape is wrong, because shape problems never get cheaper to fix later. +- APPROVE: the plan is decision-complete, correctly scoped, approach matches repo patterns, and is implementable with no required edits. Implementation starts immediately on your approval. +- REQUEST_CHANGES: anything less. Your body is the complete critique — every shape problem, every binding clarification the plan is missing (exact wire schemas, parser boundaries, error code taxonomy, side-effect boundaries, malformed-payload behavior), every nudge toward an existing repo pattern. The planner redrafts the plan once, folding your critique; you then re-review the redraft. Batch everything — a point you omit now either ships unreviewed or costs the operator fallback. -Include concise review body text. For APPROVE_WITH_REQUIRED_CHANGES, the body must contain the full set of binding requirements as plain prose or a bulleted list, since the implementer reads it directly. +Include concise review body text. For REQUEST_CHANGES, the body must contain the full critique as plain prose or a bulleted list — the planner folds it into the redraft directly, and if the redraft still fails, the operator reads it at the park. -LOW-LEVEL API CONTRACT RULE: When a binding requirement specifies a low-level contract — an exact method call, function signature, library API, or framework internal (e.g. `await request.abody()`, `Model.objects.abulk_create(...)`, a specific Django/DRF/Ninja method) — you MUST verify the framework actually supports it before making it binding. Verify by: grepping the repo for prior usage (`rg "method_name" backend/`), reading adjacent code that does similar work, or confirming against the repo's pinned dependency versions. If you cannot verify framework support, express the requirement as BEHAVIOR rather than an exact call — write "read the request body asynchronously without blocking the event loop" not "use `await request.abody()`". Specifying an unverified low-level call as binding poisons the entire implementation loop: the implementer either blindly complies and ships a runtime error, or spends multiple rounds discovering the call doesn't exist. When in doubt, state the intent and leave the implementer to pick the correct API. +LOW-LEVEL API CONTRACT RULE: When your critique specifies a low-level contract — an exact method call, function signature, library API, or framework internal (e.g. `await request.abody()`, `Model.objects.abulk_create(...)`, a specific Django/DRF/Ninja method) — you MUST verify the framework actually supports it before making it binding. Verify by: grepping the repo for prior usage (`rg "method_name" backend/`), reading adjacent code that does similar work, or confirming against the repo's pinned dependency versions. If you cannot verify framework support, express the point as BEHAVIOR rather than an exact call — write "read the request body asynchronously without blocking the event loop" not "use `await request.abody()`". An unverified low-level call in the critique poisons the redrafted plan: the implementer either blindly complies and ships a runtime error, or spends multiple rounds discovering the call doesn't exist. When in doubt, state the intent and leave the implementer to pick the correct API. -VERIFICATION FEASIBILITY & SCOPE RULE: A binding requirement is only worth writing if the implementer can actually satisfy it in the sandbox. Two failure modes deadlock the whole loop — the implementer can't win, and the evaluator re-runs it every round until the revision cap escalates to a human: +VERIFICATION FEASIBILITY & SCOPE RULE: A requirement is only worth writing if the implementer can actually satisfy it in the sandbox. Two failure modes deadlock the whole loop — the implementer can't win, and the evaluator re-runs it every round until the revision cap escalates to a human: - **Un-runnable mandatory verification.** Before promoting a verification command (test suite, production build, typecheck) to mandatory, consider whether it can even run in the sandbox: right runtime major (Node/Python), deps installed, no private-registry or network it lacks. If you can't be confident it executes, frame it as "run and report results if the command executes; otherwise report the exact command and blocker as not_run" — never a hard gate the box provably cannot build. A mandatory check the sandbox can't execute blocks the PR forever with no path forward. -- **Mandate-vs-forbid contradiction.** Never make a requirement mandatory while another requirement — or your own out-of-scope guard — forbids the only change that would satisfy it. If passing the frontend build would need a Node bump but you also forbid touching dependencies or the runtime, you've written an unsatisfiable contract. Resolve it one of three ways: allow the enabling change, drop the mandate, or hand it to the operator as an explicit out-of-scope note. Do not ship both halves of the contradiction. +- **Mandate-vs-forbid contradiction.** Never make one requirement mandatory while another — or your own out-of-scope guard — forbids the only change that would satisfy it. If passing the frontend build would need a Node bump but you also forbid touching dependencies or the runtime, you've written an unsatisfiable contract. Resolve it one of three ways: allow the enabling change, drop the mandate, or hand it to the operator as an explicit out-of-scope note. Do not ship both halves of the contradiction. Scale rigor to the change. A lint-only or single-file ticket does not warrant promoting the entire test suite plus a production build of unrelated surfaces to mandatory verification. Require verification proportional to what the diff actually touches and to what the sandbox can run; push broader hardening to a follow-up ticket rather than gating a small change on a full-suite green the environment can't even produce. - -ASSIGNEE RESOLUTION — the `assignee_github_login` schema field. Resolve the ticket -assignee's GitHub login via the github MCP from their name -`{{ build.task_owner_name or "(unknown)" }}` or email -`{{ build.task_owner_email or "(unknown)" }}` (user search; pick the -candidate whose profile clearly matches). Report the login string, or `null` when -nothing resolves convincingly — never guess. Druks uses it to request their -review at the parks that await a human; do not request reviewers yourself. diff --git a/backend/tests/test_author_surface.py b/backend/tests/test_author_surface.py index f25003f..50ee0ec 100644 --- a/backend/tests/test_author_surface.py +++ b/backend/tests/test_author_surface.py @@ -13,6 +13,8 @@ "AgentCallStatus", "FatalError", "Gate", + "Journal", + "OperatorReply", "Run", "RunState", "SubjectActivity", diff --git a/backend/tests/test_build_durable.py b/backend/tests/test_build_durable.py index 784c6f5..227d9cc 100644 --- a/backend/tests/test_build_durable.py +++ b/backend/tests/test_build_durable.py @@ -103,9 +103,15 @@ async def _wait(engine, wfid, predicate, timeout=20.0): raise AssertionError("timed out") -def _stub(monkeypatch, rt): +def _stub(monkeypatch, rt, *, plan_approval="human", auto_dispatch=False): import druks.build.workflows as m - from druks.build.contracts import CodeReviewOutput, EvaluationOutput, PlanData + from druks.build.contracts import ( + CodeReviewOutput, + EvaluationOutput, + ImplementationOutput, + PlanData, + ReviewOutput, + ) from druks.build.enums import EvaluationVerdict, ReviewDecision from druks.build.policy import Gates, RepoPolicy @@ -124,14 +130,16 @@ async def _noop(*a, **k): async def _policy_and_profile(self): policy = RepoPolicy( - gates=Gates(plan_approval="human", implementation_approval="human"), + gates=Gates(plan_approval=plan_approval, implementation_approval="human"), on_approval="merge", ) return {"policy": policy.model_dump(mode="json"), "profile": {}} async def _settings(self): return flow.Settings( - auto_dispatch_on_plan_approval=False, max_implementation_revisions=5, review_code=True + auto_dispatch_on_plan_approval=auto_dispatch, + max_implementation_revisions=5, + review_code=True, ) monkeypatch.setattr(flow, "_load_policy_and_profile", _policy_and_profile) @@ -139,17 +147,30 @@ async def _settings(self): # The agent execution is faked per agent BELOW the step wrapper (_run, not # __call__), so every call still memoizes through DBOS exactly like prod — - # the recovery test's call counts prove replay skips them. Pass-through - # stages (generate_plan, review_code) return their stub as the step result, - # so those stubs are the real domain models; contract-consuming stages take - # attribute lookalikes. Returns the invocation log (agent ids, in order). + # the recovery test's call counts prove replay skips them. The stubs are + # real domain models: they land on the journal, so its typed projections + # read them exactly as in prod. Returns the invocation log + # (agent ids, in order). results = { "generate_plan": PlanData(plan_markdown="p"), - "review_plan": SimpleNamespace( - decision=ReviewDecision.APPROVE, body="", assignee_github_login=None - ), - "implement": SimpleNamespace( - status="success", base_sha="a", head_sha="b", branch="agent/acme-1", pr_number=42 + "review_plan": ReviewOutput(decision=ReviewDecision.APPROVE, body=""), + "implement": ImplementationOutput.model_validate( + { + "type": "result", + "status": "success", + "base_sha": "a", + "head_sha": "b", + "commit_sha": "b", + "branch": "agent/acme-1", + "pr_number": 42, + "files_changed": [], + "acceptance_results": [], + "checks": [], + "known_risks": [], + "summary": "", + "workspace_path": "/repo", + "workspace_retention": None, + } ), "evaluate_implementation": EvaluationOutput( verdict=EvaluationVerdict.PASS, body="", findings=[], checks=[], acceptance_results=[] @@ -220,12 +241,36 @@ async def test_happy_path_to_merge(rt, monkeypatch): assert [e.type for e in events if e.type.startswith("run.")][-1] == "run.finished" -async def test_recovery_rebuilds_the_diary_without_rerunning_agents(rt, monkeypatch): - """The diary (plans, reviews, revision counts) is durable by determinism: a - crash mid-run means DBOS re-executes run() from the top on a fresh instance, - with every agent call and gate reply memoized. Kill the runtime while the run - is parked mid-_plan_phase, bring it back up, and the run must finish — with - the pre-crash agents replayed from checkpoints, never re-invoked.""" +async def test_auto_mode_machine_review_replaces_the_plan_gate(rt, monkeypatch): + """plan_approval resolves to none: the machine reviewer approves the plan and + the run reaches the work gate with no plan park — review_plan runs exactly + once, where it substitutes for the operator.""" + invoked = _stub(monkeypatch, rt, plan_approval=None, auto_dispatch=True) + + item_id = _seed_work_item(rt.engine, repo="acme/gizmo") + wfid = await rt.flow.start( + repo="acme/gizmo", + subject={"type": "work_item", "id": item_id}, + ) + + parked = await _wait( + rt.engine, + wfid, + lambda r: r.state == RunState.PENDING_INPUT and r.input_gate == "review_work", + ) + assert invoked[:2] == ["generate_plan", "review_plan"] + await parked.resume(action="approve") + done = await _wait(rt.engine, wfid, lambda r: r.state == RunState.FINISHED) + assert done.failure is None + + +async def test_recovery_rebuilds_the_journal_without_rerunning_agents(rt, monkeypatch): + """The journal is durable by determinism: a crash mid-run means DBOS + re-executes the body from the top on a fresh instance, with every agent + call and gate reply memoized through the same chokepoints. Kill the + runtime while the run is parked mid-_plan_phase, bring it back up, and the + run must finish — with the pre-crash agents replayed from checkpoints, + never re-invoked.""" from druks.durable.engine import init_dbos, launch, shutdown invoked = _stub(monkeypatch, rt) @@ -240,7 +285,8 @@ async def test_recovery_rebuilds_the_diary_without_rerunning_agents(rt, monkeypa wfid, lambda r: r.state == RunState.PENDING_INPUT and r.input_gate == "review", ) - assert invoked == ["generate_plan", "review_plan"] + # Gate mode: the operator is the reviewer — review_plan never ran. + assert invoked == ["generate_plan"] # The crash: tear the runtime down while the workflow is parked on the gate # and bring it back up. launch() recovers the pending workflow, which @@ -262,6 +308,19 @@ async def test_recovery_rebuilds_the_diary_without_rerunning_agents(rt, monkeypa lambda r: r.state == RunState.PENDING_INPUT and r.input_gate == "review", ) await parked.resume(action="approve", answers={}) + parked = await _wait( + rt.engine, + wfid, + lambda r: r.state == RunState.PENDING_INPUT and r.input_gate == "review_work", + ) + + # Second crash with an implementation on the journal: the replay back to + # this park rebuilds the typed projections, with zero agent re-invocations. + shutdown() + init_dbos() + launch() + await DBOS._configure_asyncio_thread_pool() + parked = await _wait( rt.engine, wfid, @@ -273,11 +332,10 @@ async def test_recovery_rebuilds_the_diary_without_rerunning_agents(rt, monkeypa # Replay recomposition, proven by invocation counts: the pre-crash agents # came back from checkpoints (no re-invocation), and the post-crash phase ran - # once each on the rebuilt diary — implement's revision guard, evaluate's + # once each on the rebuilt journal — implement's revision guard, evaluate's # grade, and the review_code toggle all read recomposed state. assert invoked == [ "generate_plan", - "review_plan", "implement", "evaluate_implementation", "review_code", diff --git a/backend/tests/test_build_journal.py b/backend/tests/test_build_journal.py new file mode 100644 index 0000000..a172fe3 --- /dev/null +++ b/backend/tests/test_build_journal.py @@ -0,0 +1,92 @@ +from druks.build.contracts import ImplementationOutput, PlanData, ReviewWork, TriageOutput +from druks.build.enums import HumanFeedbackAction +from druks.build.journal import BuildJournal + + +def _journal(*entries) -> BuildJournal: + journal = BuildJournal() + for entry in entries: + journal.add(entry) + return journal + + +def _implementation(status: str = "success") -> ImplementationOutput: + delivered = status == "success" + return ImplementationOutput.model_validate( + { + "type": "result", + "status": status, + "base_sha": "a" if delivered else None, + "head_sha": "b" if delivered else None, + "commit_sha": "b" if delivered else None, + "branch": "agent/eng-1" if delivered else None, + "pr_number": 7 if delivered else None, + "files_changed": [], + "acceptance_results": [], + "checks": [], + "known_risks": [], + "summary": "", + "workspace_path": "/repo", + "workspace_retention": None, + } + ) + + +def test_plan_is_the_latest_or_an_empty_stand_in(): + assert _journal().plan.plan_markdown == "" + journal = _journal(PlanData(plan_markdown="v1"), PlanData(plan_markdown="v2")) + assert journal.plan.plan_markdown == "v2" + assert journal.plan_revision == 2 + + +def test_only_shipped_deliveries_count_as_implementations(): + journal = _journal(_implementation(), _implementation(status="needs_clarification")) + assert len(journal.implementations) == 1 + assert journal.implementation_revision == 1 + last = journal.last_implementation + assert last and last.status == "success" + assert _journal().last_implementation is None + + +def test_assignee_scan_falls_through_unresolved_revisions(): + journal = _journal( + PlanData(plan_markdown="v1", assignee_github_login="alice"), + PlanData(plan_markdown="v2", assignee_github_login="bob"), + PlanData(plan_markdown="v3"), # a revision that resolved nobody + ) + assert journal.assignee_github_login == "bob" + assert _journal(PlanData()).assignee_github_login is None + + +def test_human_feedback_pairs_each_request_changes_reply_with_its_triage(): + journal = _journal( + ReviewWork(action="approve", reviewer="carol"), # never triaged, never paired + ReviewWork(action="request_changes", reviewer="alice", body="raw review text"), + TriageOutput( + action=HumanFeedbackAction.CHANGE_REQUIRED, + body="rename the flag", + question="", + implementation_instructions="rename X to Y", + ), + ReviewWork(action="request_changes", reviewer=None), + TriageOutput( + action=HumanFeedbackAction.QUESTION, + body="", + question="is the flag permanent?", + implementation_instructions="", + ), + ) + assert journal.human_feedback == [ + { + "reviewer": "alice", + "body": "rename the flag", + "question": "", + "implementation_instructions": "rename X to Y", + }, + { + "reviewer": "(triage)", + "body": "", + "question": "is the flag permanent?", + "implementation_instructions": "", + }, + ] diff --git a/backend/tests/test_build_plan_phase.py b/backend/tests/test_build_plan_phase.py index af75005..6e84d86 100644 --- a/backend/tests/test_build_plan_phase.py +++ b/backend/tests/test_build_plan_phase.py @@ -5,75 +5,187 @@ from druks.build.enums import ReviewDecision from druks.build.policy import RepoPolicy from druks.build.workflows import Build, BuildWorkflow -from druks.workflows import FatalError, ReviewReply +from druks.workflows import FatalError, OperatorReply -async def test_plan_phase_threads_free_text_into_the_next_pass(monkeypatch): - """A free-text answer and a request_changes note both reach the next plan pass - (answered_questions / operator_note) — a re-plan is never blind.""" +def _flow(*, auto_dispatch: bool = False) -> BuildWorkflow: flow = BuildWorkflow() - flow._policy = RepoPolicy() # plan_approval defaults to the human gate - flow._settings = BuildWorkflow.Settings() + # plan_approval is undeclared: the gate resolves to "none" iff auto_dispatch. + flow._policy = RepoPolicy() + flow._settings = BuildWorkflow.Settings(auto_dispatch_on_plan_approval=auto_dispatch) + return flow - plans = iter( - [ - PlanData( - plan_markdown="v1", - questions=[ - QuestionOutput( - id="q1", - prompt="Which cache?", - options=[QuestionOptionOutput(id="a", label="Redis")], - ) - ], - ), - PlanData(plan_markdown="v2"), - PlanData(plan_markdown="v3"), - ] - ) + +def _fake_plans(monkeypatch, *plans: PlanData) -> list[dict]: passes: list[dict] = [] + supply = iter(plans) async def fake_plan_agent(**kwargs): - # The call kwargs the planner's view is built from on each pass. passes.append(kwargs) - return next(plans) + return next(supply) + + monkeypatch.setattr(Build, "generate_plan", fake_plan_agent) + return passes + + +def _fake_grades(monkeypatch, *grades: ReviewOutput) -> None: + supply = iter(grades) async def fake_review_agent(): - return ReviewOutput( - decision=ReviewDecision.REQUEST_CHANGES, body="", assignee_github_login=None - ) + return next(supply) - monkeypatch.setattr(Build, "generate_plan", fake_plan_agent) monkeypatch.setattr(Build, "review_plan", fake_review_agent) + +def _no_review_agent(monkeypatch) -> None: + async def fail_review_agent(): + raise AssertionError("review_plan must not run here") + + monkeypatch.setattr(Build, "review_plan", fail_review_agent) + + +async def test_gate_mode_parks_every_plan_and_never_calls_the_reviewer(monkeypatch): + """Gate mode: generate → park, the reviewer never runs; operator answers + and notes thread into the next pass.""" + flow = _flow() + passes = _fake_plans( + monkeypatch, + PlanData( + plan_markdown="v1", + questions=[ + QuestionOutput( + id="q1", + prompt="Which cache?", + options=[QuestionOptionOutput(id="a", label="Redis")], + ) + ], + ), + PlanData(plan_markdown="v2"), + PlanData(plan_markdown="v3"), + ) + _no_review_agent(monkeypatch) + replies = iter( [ - ReviewReply( + OperatorReply( action="request_changes", answers={"q1": "memcache — redis is banned here"} ), - ReviewReply(action="request_changes", note="add a rollback section"), - ReviewReply(action="approve"), + OperatorReply(action="request_changes", note="add a rollback section"), + OperatorReply(action="approve"), ] ) - async def fake_review(*, questions=None): + async def fake_review(*, questions=None, context=""): return next(replies) flow.review = fake_review assert await flow._plan_phase() is True assert passes == [ - {"answered_questions": [], "operator_note": ""}, + {"answered_questions": [], "operator_note": "", "reviewer_notes": ""}, { "answered_questions": [ {"question": "Which cache?", "answer": "memcache — redis is banned here"} ], "operator_note": "", + "reviewer_notes": "", }, - {"answered_questions": [], "operator_note": "add a rollback section"}, + {"answered_questions": [], "operator_note": "add a rollback section", "reviewer_notes": ""}, ] +async def test_auto_mode_folds_the_critique_into_one_redraft(monkeypatch): + """Auto mode, no questions: REQUEST_CHANGES routes the critique into one + redraft (reviewer_notes), the re-review approves, and nothing parks.""" + flow = _flow(auto_dispatch=True) + passes = _fake_plans(monkeypatch, PlanData(plan_markdown="v1"), PlanData(plan_markdown="v2")) + _fake_grades( + monkeypatch, + ReviewOutput(decision=ReviewDecision.REQUEST_CHANGES, body="name the wire schema"), + ReviewOutput(decision=ReviewDecision.APPROVE, body=""), + ) + + async def fail_park(*, questions=None, context=""): + raise AssertionError("an approved auto-mode plan must not park") + + flow.review = fail_park + + assert await flow._plan_phase() is True + assert [p["reviewer_notes"] for p in passes] == ["", "name the wire schema"] + + +async def test_auto_mode_parks_after_the_bounded_redraft(monkeypatch): + """Two straight rejections exhaust the machine loop — the run parks with the + critique standing. The operator's request_changes re-arms one fresh redraft.""" + flow = _flow(auto_dispatch=True) + passes = _fake_plans( + monkeypatch, + PlanData(plan_markdown="v1"), + PlanData(plan_markdown="v2"), + PlanData(plan_markdown="v3"), + PlanData(plan_markdown="v4"), + ) + _fake_grades( + monkeypatch, + ReviewOutput(decision=ReviewDecision.REQUEST_CHANGES, body="critique-1"), + ReviewOutput(decision=ReviewDecision.REQUEST_CHANGES, body="critique-2"), + ReviewOutput(decision=ReviewDecision.REQUEST_CHANGES, body="critique-3"), + ReviewOutput(decision=ReviewDecision.APPROVE, body=""), + ) + parks: list[tuple[list, str]] = [] + replies = iter([OperatorReply(action="request_changes", note="steer left")]) + + async def fake_review(*, questions=None, context=""): + parks.append((list(questions or []), context)) + return next(replies) + + flow.review = fake_review + + assert await flow._plan_phase() is True + # One park, after the exhausted redraft, carrying the final critique. + assert parks == [([], "critique-2")] + assert [p["reviewer_notes"] for p in passes] == ["", "critique-1", "", "critique-3"] + assert [p["operator_note"] for p in passes] == ["", "", "steer left", "steer left"] + + +async def test_questions_park_in_auto_mode_too(monkeypatch): + """Open questions always park for the operator — the machine reviewer only + ever sees a question-free plan.""" + flow = _flow(auto_dispatch=True) + _fake_plans( + monkeypatch, + PlanData( + plan_markdown="v1", + questions=[QuestionOutput(id="q1", prompt="Feature flag?", options=[])], + ), + PlanData(plan_markdown="v2"), + ) + _fake_grades(monkeypatch, ReviewOutput(decision=ReviewDecision.APPROVE, body="")) + replies = iter([OperatorReply(action="request_changes", answers={"q1": "yes, behind a flag"})]) + + async def fake_review(*, questions=None, context=""): + assert questions # the park carries the open questions + return next(replies) + + flow.review = fake_review + + assert await flow._plan_phase() is True + + +async def test_cancel_at_the_plan_park_stops_the_run(monkeypatch): + flow = _flow() + _fake_plans(monkeypatch, PlanData(plan_markdown="v1")) + _no_review_agent(monkeypatch) + + async def fake_review(*, questions=None, context=""): + return OperatorReply(action="cancel") + + flow.review = fake_review + + with pytest.raises(FatalError, match="cancelled at plan review"): + await flow._plan_phase() + + async def test_needs_clarification_delivery_stops_the_run(monkeypatch): """The implementer bailing (needs_clarification) fails the run with its own reason — the stop is a workflow decision now, not a contract side effect.""" diff --git a/backend/tests/test_build_prompts.py b/backend/tests/test_build_prompts.py index ec48be4..8daea2b 100644 --- a/backend/tests/test_build_prompts.py +++ b/backend/tests/test_build_prompts.py @@ -20,13 +20,13 @@ # The kwargs the workflow passes at each template's agent call site. _CALL_KWARGS = { - "generate_plan.md": {"answered_questions": [], "operator_note": ""}, + "generate_plan.md": {"answered_questions": [], "operator_note": "", "reviewer_notes": ""}, } def _build() -> SimpleNamespace: """A stand-in BuildPromptContext exposing the fields the templates read — - identity facts faked, the journal real.""" + identity facts faked, the journal real and empty.""" return SimpleNamespace( repo="acme/widget", branch="agent/eng-1", @@ -91,11 +91,45 @@ async def test_generate_plan_prompt_quotes_operator_content(): workspace=_workspace(), answered_questions=[{"question": "Which cache?", "answer": "redis\nwith a 5m TTL"}], operator_note="Tighten the rollout.\nSplit phase 2.", + reviewer_notes="", ) assert "> redis\n > with a 5m TTL" in output assert "> Tighten the rollout.\n> Split phase 2." in output +async def test_generate_plan_prompt_quotes_the_reviewer_critique(): + output = await render_prompt( + "build/build_workflow/generate_plan.md", + build=_build(), + verification="VERIFICATION-BLOCK", + workspace=_workspace(), + answered_questions=[], + operator_note="", + reviewer_notes="Name the wire schema.\nSplit the migration.", + ) + assert "## Plan reviewer critique" in output + assert "> Name the wire schema.\n> Split the migration." in output + + +async def test_the_planner_resolves_the_assignee_not_the_reviewer(): + # Only the planner always runs, so only its prompt owns the resolution. + planner = await render_prompt( + "build/build_workflow/generate_plan.md", + build=_build(), + verification="VERIFICATION-BLOCK", + workspace=_workspace(), + **_CALL_KWARGS["generate_plan.md"], + ) + reviewer = await render_prompt( + "build/build_workflow/review_plan.md", + build=_build(), + verification="VERIFICATION-BLOCK", + workspace=_workspace(), + ) + assert "ASSIGNEE RESOLUTION" in planner + assert "assignee_github_login" not in reviewer + + @pytest.mark.parametrize("template", _OP_TEMPLATES) async def test_build_prompt_orders_the_ticket_fetch(template): """Every build agent is ordered to fetch the ticket from its source before diff --git a/backend/tests/test_contracts.py b/backend/tests/test_contracts.py index 53a80de..1bc8b8d 100644 --- a/backend/tests/test_contracts.py +++ b/backend/tests/test_contracts.py @@ -129,4 +129,13 @@ def test_ask_contracts_cap_identity_and_cardinality(): plan_markdown="m", acceptance_criteria=[], questions=[O.QuestionOutput(id=f"q{i}", prompt="p", options=[]) for i in range(9)], + assignee_github_login=None, ) + + +def test_review_output_records_no_artifact(): + # An artifact would displace the plan as the parked ask's document. + from druks.build.enums import ReviewDecision + + grade = O.ReviewOutput(decision=ReviewDecision.REQUEST_CHANGES, body="name the wire schema") + assert grade.get_artifact() == {} diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index ea54a54..d6020da 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -82,18 +82,18 @@ class AgentFlow(Workflow): async def run(self, repo: str) -> None: decision = await self.DECIDER(body="x") # run()'s whole body is one step, so this agent call is in-step and - # must never land on the record — on the live or the replay pass. - SINK.append(f"instep-record:{len(self.record.list(Decision))}") + # must never land on the journal — on the live or the replay pass. + SINK.append(f"instep-journal:{len(self.journal.filter(Decision))}") if decision.action == "stop": raise FatalError("stopped by agent") class AgentBodyFlow(Workflow): - # The record chokepoint: in run_multistep the agent call is body-level — - # its own step — so the platform records the domain value it returns. + # In run_multistep the agent call is body-level — its own step — so + # the platform journals the domain value it returns. async def run_multistep(self, repo: str) -> None: decision = await AgentFlow.DECIDER(body="x") - assert self.record.latest(Decision) is decision - SINK.append(f"body-record:{len(self.record.list(Decision))}:{decision.action}") + assert self.journal.latest(Decision) is decision + SINK.append(f"body-journal:{len(self.journal.filter(Decision))}:{decision.action}") class RecordFeedback(Workflow): @step @@ -125,8 +125,8 @@ async def run_multistep(self) -> None: SINK.append(f"round1:{first.action}") second = await Approve.wait() SINK.append(f"round2:{second.action}") - replies = [reply.action for reply in self.record.list(Approve)] - SINK.append(f"gate-record:{replies}") + replies = [reply.action for reply in self.journal.filter(Approve)] + SINK.append(f"gate-journal:{replies}") class ConfirmFlow(Workflow): async def run_multistep(self) -> None: @@ -390,8 +390,8 @@ async def test_duplicate_replies_to_one_round_collapse(rt): await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED) assert "round2:second" in SINK assert "round2:duplicate" not in SINK - # Both replies landed on the run record, in reply order. - assert "gate-record:['first', 'second']" in SINK + # Both replies landed on the journal, in reply order. + assert "gate-journal:['first', 'second']" in SINK async def test_fail_branch(rt): @@ -508,7 +508,7 @@ async def test_run_agent_step(rt, monkeypatch): wfid = await rt.AgentFlow.start(subject=None, repo="owner/app") failed = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FAILED) assert failed.failure == "stopped by agent" - assert "instep-record:0" in SINK # an agent call inside run()'s step: never recorded + assert "instep-journal:0" in SINK # an agent call inside run()'s step: never recorded assert seen and seen[0]["artifact_dir"].name == f"run-{wfid}" assert seen[0]["agent"] == "decider" session = get_session(rt.engine) @@ -524,10 +524,10 @@ async def test_run_agent_step(rt, monkeypatch): assert pinned == [0] # connection released while the agent runs -async def test_body_level_agent_output_lands_on_the_record(rt, monkeypatch): - """The record chokepoint: a run_multistep body-level agent call appends the - domain value the body receives — AgentBodyFlow asserts identity in-body and - sinks the projection.""" +async def test_body_level_agent_output_lands_on_the_journal(rt, monkeypatch): + """A run_multistep body-level agent call lands the domain value the body + receives on the journal — AgentBodyFlow asserts identity in-body and sinks + the projection.""" seen: list[dict] = [] pinned: list[int] = [] monkeypatch.setattr( @@ -537,7 +537,7 @@ async def test_body_level_agent_output_lands_on_the_record(rt, monkeypatch): wfid = await rt.AgentBodyFlow.start(subject=None, repo="owner/app") await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED) - assert "body-record:1:ship" in SINK + assert "body-journal:1:ship" in SINK async def test_task_enqueue(rt): diff --git a/backend/tests/test_journal.py b/backend/tests/test_journal.py new file mode 100644 index 0000000..345c0e2 --- /dev/null +++ b/backend/tests/test_journal.py @@ -0,0 +1,60 @@ +import pytest +from druks.workflows import Journal, OperatorReply +from pydantic import BaseModel + + +class Finding(BaseModel): + status: str + title: str = "" + + +class Grade(BaseModel): + decision: str + + +def _journal() -> Journal: + journal = Journal() + journal.add(Finding(status="success", title="a")) + journal.add(Grade(decision="approve")) + journal.add(Finding(status="failed", title="b")) + journal.add(Finding(status="success", title="c")) + return journal + + +def test_filter_selects_by_contract_type_in_call_order(): + journal = _journal() + assert [finding.title for finding in journal.filter(Finding)] == ["a", "b", "c"] + assert [grade.decision for grade in journal.filter(Grade)] == ["approve"] + + +def test_latest_is_the_newest_match_or_none(): + journal = _journal() + latest = journal.latest(Finding) + assert latest and latest.title == "c" + assert Journal().latest(Finding) is None + + +def test_filters_are_flat_anded_equality(): + journal = _journal() + assert [f.title for f in journal.filter(Finding, status="success")] == ["a", "c"] + assert [f.title for f in journal.filter(Finding, status="success", title="a")] == ["a"] + assert journal.latest(Finding, status="missing") is None + assert Journal().filter(Finding, status="success") == [] + + +def test_a_filter_typo_raises_when_entries_scan(): + with pytest.raises(AttributeError): + _journal().filter(Finding, verdict="success") + + +def test_review_reply_validates_the_resume_wire_shape(): + # The resume endpoint's payload — {action, answers, note} — validates + # unchanged; answers and note default for partial senders. + reply = OperatorReply.model_validate( + {"action": "request_changes", "answers": {"q1": "redis"}, "note": "why q1"} + ) + assert reply.action == "request_changes" + assert (reply.answers, reply.note) == ({"q1": "redis"}, "why q1") + assert OperatorReply.model_validate({"action": "approve"}).answers == {} + # The recv topic stays the wire's "review", not the class-name derivation. + assert OperatorReply.topic == "review" diff --git a/backend/tests/test_notifications_durable.py b/backend/tests/test_notifications_durable.py index 3d3d5ac..bc4d350 100644 --- a/backend/tests/test_notifications_durable.py +++ b/backend/tests/test_notifications_durable.py @@ -16,7 +16,7 @@ from druks.notifications.outbox import notifications_queue, send_notification from druks.notifications.services import respond_to_notification from druks.user_settings.models import UserSettings -from druks.workflows import Gate, Run, Workflow +from druks.workflows import Gate, OperatorReply, Run, Workflow from fastapi.testclient import TestClient from pydantic import BaseModel, Field from sqlalchemy import create_engine, select, text @@ -49,7 +49,7 @@ class _Question(BaseModel): # What the in-app reviews resumed with — the respond round-trip asserts the # payload arrived in the workflow verbatim. -_REVIEW_REPLIES: list[dict] = [] +_REVIEW_REPLIES: list[OperatorReply] = [] def _build_park_flows(): @@ -662,7 +662,7 @@ async def test_respond_round_trip_finishes_the_run(rt, deliver_spy): assert first == "ok" await _wait_run(rt, workflow_id, lambda run: run.state == RunState.FINISHED) - assert {"action": "approve", "answers": {"q1": "postgres"}, "note": ""} in _REVIEW_REPLIES + assert OperatorReply(action="approve", answers={"q1": "postgres"}) in _REVIEW_REPLIES assert _notifications_for_run(rt, workflow_id)[0].state == "acknowledged" # Sequential second answer: the acknowledged fast-path rejects it, and the diff --git a/backend/tests/test_run_record.py b/backend/tests/test_run_record.py deleted file mode 100644 index 979ad6b..0000000 --- a/backend/tests/test_run_record.py +++ /dev/null @@ -1,60 +0,0 @@ -import pytest -from druks.workflows import ReviewReply, RunRecord -from pydantic import BaseModel - - -class Finding(BaseModel): - status: str - title: str = "" - - -class Grade(BaseModel): - decision: str - - -def _record() -> RunRecord: - record = RunRecord() - record._append(Finding(status="success", title="a")) - record._append(Grade(decision="approve")) - record._append(Finding(status="failed", title="b")) - record._append(Finding(status="success", title="c")) - return record - - -def test_selection_is_by_contract_type_in_call_order(): - record = _record() - assert [finding.title for finding in record.list(Finding)] == ["a", "b", "c"] - assert [grade.decision for grade in record.list(Grade)] == ["approve"] - - -def test_latest_is_the_newest_match_or_none(): - record = _record() - latest = record.latest(Finding) - assert latest and latest.title == "c" - assert RunRecord().latest(Finding) is None - - -def test_filters_are_flat_anded_equality(): - record = _record() - assert [f.title for f in record.list(Finding, status="success")] == ["a", "c"] - assert [f.title for f in record.list(Finding, status="success", title="a")] == ["a"] - assert record.latest(Finding, status="missing") is None - assert RunRecord().list(Finding, status="success") == [] - - -def test_a_filter_typo_raises_when_entries_scan(): - with pytest.raises(AttributeError): - _record().list(Finding, verdict="success") - - -def test_review_reply_validates_the_resume_wire_shape(): - # The resume endpoint's payload — {action, answers, note} — validates - # unchanged; answers and note default for partial senders. - reply = ReviewReply.model_validate( - {"action": "request_changes", "answers": {"q1": "redis"}, "note": "why q1"} - ) - assert reply.action == "request_changes" - assert (reply.answers, reply.note) == ({"q1": "redis"}, "why q1") - assert ReviewReply.model_validate({"action": "approve"}).answers == {} - # The recv topic stays the wire's "review", not the class-name derivation. - assert ReviewReply.topic == "review" diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md index 3d533c7..1dea069 100644 --- a/docs/writing-an-extension.md +++ b/docs/writing-an-extension.md @@ -146,41 +146,42 @@ visible by comparison. Runs with no account anywhere (crons, background work) run as the system account. Resuming a parked run keeps its original attribution; the person clicking Resume never becomes the payer. -### The run record +### The journal -The platform records the run's typed history for you. Every body-level agent -call appends the domain value the body receives (the post-`to_result()` -output), and every gate reply appends its validated `Gate` instance — -`review()`'s `ReviewReply` included — in call order, on `self.record`. Project -working state off it instead of hand-maintaining collections: +Druks keeps a journal of each run's typed values: every body-level agent call +and every gate reply lands in it automatically, in call order. Add your own +values with `self.journal.add()`; read them back by contract type: ```python -self.record.list(PlanData) # ordered, all of them -self.record.latest(PlanData) # newest or None -self.record.list(ImplementationOutput, status="success") # equality kwargs, ANDed -self.record.latest(ReviewOutput, decision=ReviewDecision.APPROVE) -self.record.list(ReviewWork) # gate replies: instances of the Gate class +self.journal.filter(PlanData) # all entries, oldest first +self.journal.latest(PlanData) # newest, or None +self.journal.filter(ImplementationOutput, status="success") # keyword filters: ANDed equality +self.journal.filter(ReviewWork) # gate replies, by their Gate class ``` -Selection is by contract type, never by producer — two agents producing the -same noun land in one projection. Filters are flat `attr=value` equality, -ANDed, the `@subscribe`-filter idiom. That is the whole API: no `after=`, no -windows, no ordinals. If a projection can't be written with `list`/`latest`, -the value's meaning depends on record position — it isn't complete at birth; -fix the contract, not the selector. - -The record is rebuilt, not stored: recovery re-runs the body and feeds the -identical memoized values through the identical chokepoints, so it rebuilds -bit-for-bit. There is no second persistence path. - -The boundary: the record covers `run_multistep()` body-level calls. An agent -call inside a `@step` — or anywhere in a `run()` body, which is one step -whole — is never recorded, consistently on the live and replay passes. A -`@step`'s own returns are plain dicts and rows with no contract type to select -by; hold those in local variables or plain instance attributes, mutated only -from body code. Never mutate such state from inside a `@step`: a completed -step is skipped on replay, so a write made there silently vanishes from the -rebuilt state. +Subclass `Journal` to name your projections, and declare it on the workflow: + +```python +class SweepJournal(Journal): + @property + def findings(self) -> list[FindingData]: + return self.filter(FindingData) + + +class Sweep(Workflow): + journal_class = SweepJournal +``` + +The journal survives crashes without being stored: recovery re-runs the body +with every durable call memoized, so the same entries land in the same order. + +Two rules: + +- Only body-level calls are journaled. An agent call inside a `@step` — or in + a `run()` body, which is one big step — never lands there; keep that state + in local variables. +- Never mutate body-held state inside a `@step`: a completed step is skipped + on replay, so the write disappears. ### Schedules and settings diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 14516fe..7353342 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -97,6 +97,8 @@ export interface InputRequest { controls?: string[] questions?: AskQuestion[] artifact_id?: string | null + /** Workflow-declared prose rendered beside the reviewed document. */ + context?: string } // A call's renderable output, fetched to render inside an in-app review. diff --git a/frontend/src/extensions/build/WorkItemPage.tsx b/frontend/src/extensions/build/WorkItemPage.tsx index 0bd722e..858c735 100644 --- a/frontend/src/extensions/build/WorkItemPage.tsx +++ b/frontend/src/extensions/build/WorkItemPage.tsx @@ -653,6 +653,12 @@ function InAppReview({ runId, ask }: { runId: string; ask: InputRequest }) { return (
+ {ask.context && ( +
+
Plan reviewer critique
+ +
+ )} {artifact.data && (
{artifact.data.title}