The journal: one add door, typed projections; plan review runs iff its verdict routes

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:

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):

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"
 

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)
         ]