diff --git a/backend/druks/build/journal.py b/backend/druks/build/journal.py new file mode 100644 index 0000000..ab76859 --- /dev/null +++ b/backend/druks/build/journal.py @@ -0,0 +1,83 @@ +from dataclasses import dataclass, field + +from druks.build.contracts import ( + EvaluationOutput, + HumanFeedback, + ImplementationOutput, + PlanData, + ReviewOutput, +) +from druks.build.enums import ReviewDecision + + +@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 + + @property + def plan(self) -> PlanData: + return self.plans[-1].plan if self.plans else PlanData() + + @property + def last_implementation(self) -> ImplementationOutput | None: + return self.implementations[-1] if self.implementations else None + + @property + def plan_revision(self) -> int: + return len(self.plans) + + @property + def implementation_revision(self) -> int: + return len(self.implementations) + + @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 + + 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 [] + return [ + review + for review in current + if review.decision == ReviewDecision.APPROVE_WITH_REQUIRED_CHANGES + and review.body.strip() + ] diff --git a/backend/druks/build/prompt_context.py b/backend/druks/build/prompt_context.py index d5f5e5f..aba2612 100644 --- a/backend/druks/build/prompt_context.py +++ b/backend/druks/build/prompt_context.py @@ -1,18 +1,13 @@ from dataclasses import dataclass -from druks.build.contracts import ( - AcceptanceCriterionOutput, - EvaluationOutput, - HumanFeedback, - ReviewOutput, -) +from druks.build.journal import BuildJournal from druks.build.models import ProjectRepo @dataclass(frozen=True) class BuildPromptContext: - """What build's prompt templates render against — a flat snapshot of the run's - state, assembled once per agent call by ``BuildWorkflow.get_prompt_context``. + """What build's prompt templates render against — the run's identity facts + plus its journal, assembled per agent call by ``BuildWorkflow.get_prompt_context``. The templates read ``build.`` and nothing else off the run, so this is the whole contract between the workflow and its prompts; the workflow itself stays free of template accessors. @@ -27,15 +22,5 @@ class BuildPromptContext: issue_number: int | None task_owner_name: str | None task_owner_email: str | None - # Where the run stands in its plan / implement loops. - plan_revision: int - implementation_revision: int - finalized_base_sha: str | None - finalized_pr_sha: str | None - # The current plan the downstream agents build on. - current_plan: str - acceptance_criteria: list[AcceptanceCriterionOutput] - reviewer_requirements: list[ReviewOutput] - implementation_reviews: list[EvaluationOutput] - human_feedback: list[HumanFeedback] related_repos: list[ProjectRepo] + journal: BuildJournal diff --git a/backend/druks/build/subscribers.py b/backend/druks/build/subscribers.py index ef38668..8600f3f 100644 --- a/backend/druks/build/subscribers.py +++ b/backend/druks/build/subscribers.py @@ -8,7 +8,7 @@ from druks.build.models import Project, ProjectRepo, WorkItem from druks.build.policy import RepoPolicy from druks.build.scoping.workflows import Scope -from druks.build.workflows import BuildWorkflow, Profile, _delete_branch +from druks.build.workflows import BuildWorkflow, Profile from druks.core.apis.exceptions import GitHubAppNotConfiguredError, GitHubAppNotInstalledError from druks.core.apis.github import get_github_client from druks.settings import load_settings @@ -137,8 +137,8 @@ async def _observe_external_close(*, repo: str, pr_number: int, work_item: WorkI # failure must not strand the ticket — its reset below still runs. try: if (await RepoPolicy.resolve(repo)).delete_branch: - await _delete_branch(repo, work_item.branch) - except Exception: # noqa: BLE001 — cleanup only, like _delete_branch itself + await get_github_client(load_settings()).delete_branch(repo, work_item.branch) + except Exception: # noqa: BLE001 — cleanup only logger.warning("Skipped branch cleanup for %s#%s.", repo, pr_number, exc_info=True) # The attempt was abandoned, not the ticket: send it back to the # provider's resting pool rather than stranding it in In Progress. diff --git a/backend/druks/build/workflows.py b/backend/druks/build/workflows.py index 9a67e16..d44a2af 100644 --- a/backend/druks/build/workflows.py +++ b/backend/druks/build/workflows.py @@ -5,13 +5,7 @@ from pydantic import BaseModel, Field from druks.accounts.models import Account -from druks.build.contracts import ( - EvaluationOutput, - HumanFeedback, - ImplementationOutput, - PlanData, - ReviewOutput, -) +from druks.build.contracts import HumanFeedback, ImplementationOutput from druks.build.enums import ( EvaluationVerdict, HumanFeedbackAction, @@ -33,6 +27,7 @@ from druks.workflows import FatalError, Gate, Workflow, step from .extension import Build +from .journal import BuildJournal from .policy import RepoPolicy from .prompt_context import BuildPromptContext from .workspace import RepoWorkspace @@ -118,17 +113,7 @@ class Settings(BaseModel): def __init__(self) -> None: super().__init__() - # The run's working memory. Durable by determinism: a recovery re-runs - # run() from the top with every agent call and gate reply memoized, so - # these rebuild to exactly what the live pass held. - self._plans: list[PlanData] = [] - self._plan_reviews: list[ReviewOutput] = [] - # Where _plan_reviews stood at the latest plan draft, so - # reviewer_requirements only spans reviews of the current plan. - self._reviews_at_plan = 0 - self._implementation_reviews: list[EvaluationOutput] = [] - self._implementation_results: list[ImplementationOutput] = [] - self._human_feedback: list[HumanFeedback] = [] + self._journal = BuildJournal() @classmethod async def dispatch( @@ -239,8 +224,6 @@ async def get_workspace_kwargs(self, sandbox: "Sandbox") -> dict[str, Any]: } async def get_prompt_context(self, **context: Any) -> dict[str, Any]: - last = self._last_implement - plan = self.plan prompt_context = BuildPromptContext( repo=self.input.repo, branch=self.branch, @@ -250,16 +233,8 @@ async def get_prompt_context(self, **context: Any) -> dict[str, Any]: issue_number=self.input.issue_number, task_owner_name=self.input.task_owner_name, task_owner_email=self.input.task_owner_email, - plan_revision=len(self._plans), - implementation_revision=len(self._implementation_results), - finalized_base_sha=last.base_sha if last else None, - finalized_pr_sha=last.head_sha if last else None, - current_plan=plan.plan_markdown, - acceptance_criteria=plan.acceptance_criteria, - reviewer_requirements=self._reviewer_requirements(), - implementation_reviews=list(self._implementation_reviews), - human_feedback=list(self._human_feedback), related_repos=self._related_repos(), + journal=self._journal, ) return { "verification": await self._policy.verification_block( @@ -291,11 +266,13 @@ async def _plan_phase(self) -> bool: answered: list[dict[str, str]] = [] note = "" while True: - plan = await self.generate_plan(answered, note) + 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 = await self.review_plan() + 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" @@ -312,7 +289,7 @@ async def _plan_phase(self) -> bool: async def _implement_phase(self) -> None: while True: await self.implement() - evaluation = await self.evaluate() + evaluation = self._journal.add_evaluation(await Build.evaluate_implementation()) if evaluation.verdict == EvaluationVerdict.PASS: if self._settings.review_code: await Build.review_code() @@ -320,7 +297,7 @@ async def _implement_phase(self) -> None: return continue if evaluation.verdict == EvaluationVerdict.FAIL and ( - len(self._implementation_results) < self._settings.max_implementation_revisions + self._journal.implementation_revision < self._settings.max_implementation_revisions ): continue if await self._work_gate(): @@ -340,7 +317,7 @@ async def _work_gate(self) -> bool: if decision.action == "request_changes": return await self._triage(decision) if decision.action == "revise_contract": - await self.revise_contract() + self._journal.add_plan(await Build.revise_contract()) return False if decision.action == "cancel": await self._push_ticket_status(SemanticStatus.CANCELED) @@ -356,36 +333,19 @@ async def _approved_work(self) -> bool: return True async def _triage(self, decision: ReviewWork) -> bool: - feedback = await self.triage_feedback(decision) - self._human_feedback.append(feedback) + feedback = self._journal.add_feedback(await self.triage_feedback(decision)) if feedback.triage_action == HumanFeedbackAction.CHANGE_REQUIRED: return False # loop → implement if feedback.triage_action == HumanFeedbackAction.CONTRACT_CHANGE_REQUIRED: - await self.revise_contract() + self._journal.add_plan(await Build.revise_contract()) return False if feedback.triage_action == HumanFeedbackAction.CLOSE: raise FatalError("closed at human triage") # NO_CHANGE / QUESTION → re-park return await self._work_gate() - # The phase methods are plain workflow-body code: the agent calls inside them - # memoize themselves, so on a recovery the pure parts re-run deterministically - # and rebuild the in-memory diary (set_state included — it is body-only). Only - # side-effecting IO beyond an agent call (GitHub writes, child starts, event - # records) keeps its own @step. - async def generate_plan(self, answered: list[dict[str, str]], note: str) -> PlanData: - return self._add_plan( - await Build.generate_plan(answered_questions=answered, operator_note=note) - ) - - async def review_plan(self) -> ReviewOutput: - grade = await Build.review_plan() - self._plan_reviews.append(grade) - return grade - - async def revise_contract(self) -> PlanData: - return self._add_plan(await Build.revise_contract()) - + # 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. async def implement(self) -> ImplementationOutput: delivery = await Build.implement() # A bail is a stop, not a result: the implementer hit a contradiction in the @@ -397,13 +357,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) - self._implementation_results.append(delivery) - return delivery - - async def evaluate(self) -> EvaluationOutput: - review = await Build.evaluate_implementation() - self._implementation_reviews.append(review) - return review + return self._journal.add_implementation(delivery) async def triage_feedback(self, decision: ReviewWork) -> HumanFeedback: parsed = await Build.triage_human_feedback() @@ -424,12 +378,15 @@ async def merge(self) -> None: return await github.squash_merge_pull_request(self.input.repo, self.pr_number) if self._policy.delete_branch: - await _delete_branch(self.input.repo, self.branch) - - def _add_plan(self, plan: PlanData) -> PlanData: - self._plans.append(plan) - self._reviews_at_plan = len(self._plan_reviews) - return plan + try: + await github.delete_branch(self.input.repo, self.branch) + except Exception: # noqa: BLE001 — cleanup only + logger.warning( + "Could not delete branch %s on %s.", + self.branch, + self.input.repo, + exc_info=True, + ) # The provisioned branch + PR, published by the first implement — None until then # (planning runs against the default branch, and there is no PR to point at). @@ -441,36 +398,6 @@ def branch(self) -> str | None: def pr_number(self) -> int | None: return getattr(self.state, "pr_number", None) - @property - def plan(self) -> PlanData: - return self._plans[-1] if self._plans else PlanData() - - @property - def plan_reviews(self) -> list[ReviewOutput]: - return list(self._plan_reviews) - - @property - def assignee_github_login(self) -> str | None: - for review in reversed(self.plan_reviews): - if review.assignee_github_login: - return review.assignee_github_login - return None - - @property - def _last_implement(self) -> ImplementationOutput | None: - return self._implementation_results[-1] if self._implementation_results else None - - # Prompt-context projections — computed for BuildPromptContext, not workflow API. - 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. - return [ - review - for review in self._plan_reviews[self._reviews_at_plan :] - if review.decision == ReviewDecision.APPROVE_WITH_REQUIRED_CHANGES - and review.body.strip() - ] - def _related_repos(self) -> list[ProjectRepo]: # The project's sibling repos (the prompt reads full_name + purpose). target = (self.input.repo or "").strip().lower() @@ -497,32 +424,30 @@ 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.assignee_github_login - if not login or not self.input.repo or not self.pr_number: - return - try: - await get_github_client(load_settings()).request_pull_request_reviewers( - self.input.repo, self.pr_number, [login] - ) - except Exception: # noqa: BLE001 — a missed ping must not fail the park - logger.warning( - "could not request review from %s on %s#%s", - login, - self.input.repo, - self.pr_number, - ) + 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( + self.input.repo, self.pr_number, [login] + ) + except Exception: # noqa: BLE001 — a missed ping must not fail the park + logger.warning( + "could not request review from %s on %s#%s", + login, + self.input.repo, + self.pr_number, + ) async def _set_pr_draft(self, *, draft: bool) -> None: - if not self.input.repo or not self.pr_number: - return - try: - await get_github_client(load_settings()).set_pull_request_draft_state( - self.input.repo, self.pr_number, draft=draft - ) - except Exception: # noqa: BLE001 — a draft merge fails loudly anyway - logger.warning( - "Could not set draft=%s on %s#%s.", draft, self.input.repo, self.pr_number - ) + if self.input.repo and self.pr_number: + try: + await get_github_client(load_settings()).set_pull_request_draft_state( + self.input.repo, self.pr_number, draft=draft + ) + except Exception: # noqa: BLE001 — a draft merge fails loudly anyway + logger.warning( + "Could not set draft=%s on %s#%s.", draft, self.input.repo, self.pr_number + ) class Profile(Workflow): @@ -583,12 +508,3 @@ async def get_prompt_context(self, **context: Any) -> dict[str, Any]: ], **await super().get_prompt_context(**context), } - - -async def _delete_branch(repo: str, branch: str | None) -> None: - if not branch: - return - try: - await get_github_client(load_settings()).delete_branch(repo, branch) - except Exception: # noqa: BLE001 — cleanup only - logger.warning("Could not delete branch %s on %s.", branch, repo, exc_info=True) diff --git a/backend/templates/prompts/build/build_workflow/_header.md b/backend/templates/prompts/build/build_workflow/_header.md index db64baf..e06170a 100644 --- a/backend/templates/prompts/build/build_workflow/_header.md +++ b/backend/templates/prompts/build/build_workflow/_header.md @@ -14,13 +14,11 @@ **MANDATORY FIRST ACTION — fetch the ticket. This is not a suggestion.** Your very first tool call MUST be to fetch `{{ build.ticket_ref }}` from {{ build.source | default('the tracker', true) | capitalize }} using your available tools, then read its full description and **every** comment before you read the codebase, write a plan, edit a file, or emit any output. Do not begin from the ticket reference, title, or the rendered plan alone — those are derived; the ticket and its operator comments are the binding source of truth, and frequently carry exact decisions you must honor verbatim (its description may also end with a `# Druks scope brief` section summarizing problem, scope, and acceptance criteria). The ONLY acceptable reason to proceed without the ticket's full text is a genuine tool failure, which you must report as a blocker — never guess or fabricate the requirements. If the source materially contradicts a plan or acceptance criteria rendered below, surface the conflict rather than silently proceeding. {% endif %} -- **Plan revision:** {{ build.plan_revision }} -- **Implementation revision:** {{ build.implementation_revision }}{% if build.implementation_revision == 0 %} (first attempt){% endif %} -{% if build.finalized_base_sha %} -- **base_sha:** `{{ build.finalized_base_sha }}` -{% endif %} -{% if build.finalized_pr_sha %} -- **head_sha:** `{{ build.finalized_pr_sha }}` +- **Plan revision:** {{ build.journal.plan_revision }} +- **Implementation revision:** {{ build.journal.implementation_revision }}{% if build.journal.implementation_revision == 0 %} (first attempt){% endif %} +{% if build.journal.last_implementation %} +- **base_sha:** `{{ build.journal.last_implementation.base_sha }}` +- **head_sha:** `{{ build.journal.last_implementation.head_sha }}` {% endif %} ### Workspace paths (inside this sandbox VM) @@ -28,16 +26,16 @@ - ``workspace_root``: `{{ workspace.workspace_root }}` — clone related repos you need as ``workspace_root/related/`` {{ verification }} -{% if build.current_plan %} +{% if build.journal.plan.plan_markdown %} ## Current plan -{{ build.current_plan }} +{{ build.journal.plan.plan_markdown }} {% endif %} -{% if build.acceptance_criteria %} +{% if build.journal.plan.acceptance_criteria %} ## Acceptance criteria -{% for ac in build.acceptance_criteria %} +{% for ac in build.journal.plan.acceptance_criteria %} ### {{ ac.id }} **Description:** {{ ac.description.strip() }} @@ -48,22 +46,22 @@ {% endif %} {% endfor %} {% endif %} -{% if build.reviewer_requirements %} +{% 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.reviewer_requirements %} +{% for req in build.journal.reviewer_requirements() %} ### Requirement {{ loop.index }} {{ req.body.strip() }} {% endfor %} {% endif %} -{% if build.implementation_reviews %} +{% if build.journal.evaluations %} ## Prior implementation review -{% for review in build.implementation_reviews %} +{% for review in build.journal.evaluations %} ### Round {{ loop.index }} — verdict: {{ review.verdict.value if review.verdict.value is defined else review.verdict }} {% if review.body %} @@ -72,10 +70,10 @@ These are binding plan clarifications from the plan reviewer. Treat them as part {% endif %} {% endfor %} {% endif %} -{% if build.human_feedback %} +{% if build.journal.human_feedback %} ## Human feedback -{% for fb in build.human_feedback %} +{% for fb in build.journal.human_feedback %} ### {{ fb.reviewer }} — status: {{ fb.status }} {% if fb.body %} diff --git a/backend/tests/test_build_prompts.py b/backend/tests/test_build_prompts.py index 0ee993d..ec48be4 100644 --- a/backend/tests/test_build_prompts.py +++ b/backend/tests/test_build_prompts.py @@ -4,6 +4,7 @@ import pytest from druks.build import workflows as build_workflows +from druks.build.journal import BuildJournal from druks.build.prompt_context import BuildPromptContext from druks.prompts import render_prompt @@ -17,15 +18,15 @@ "triage_human_feedback.md", ] -# The kwargs the workflow passes at each agent's call site — top-level template -# names, beside the standing ``build`` context. +# The kwargs the workflow passes at each template's agent call site. _CALL_KWARGS = { "generate_plan.md": {"answered_questions": [], "operator_note": ""}, } def _build() -> SimpleNamespace: - """A stand-in BuildPromptContext exposing the fields the templates read.""" + """A stand-in BuildPromptContext exposing the fields the templates read — + identity facts faked, the journal real.""" return SimpleNamespace( repo="acme/widget", branch="agent/eng-1", @@ -35,16 +36,8 @@ def _build() -> SimpleNamespace: issue_number=None, task_owner_name=None, task_owner_email=None, - plan_revision=1, - implementation_revision=0, - finalized_base_sha=None, - finalized_pr_sha=None, - current_plan=None, - acceptance_criteria=[], - reviewer_requirements=[], - implementation_reviews=[], - human_feedback=[], related_repos=[], + journal=BuildJournal(), ) diff --git a/backend/tests/test_webhooks_pull_request.py b/backend/tests/test_webhooks_pull_request.py index fb2f88f..b19cac5 100644 --- a/backend/tests/test_webhooks_pull_request.py +++ b/backend/tests/test_webhooks_pull_request.py @@ -259,7 +259,9 @@ async def _fetch(*, repo, path): async def _record(repo, branch): deleted.append((repo, branch)) - monkeypatch.setattr(webhooks_mod, "_delete_branch", _record) + monkeypatch.setattr( + webhooks_mod, "get_github_client", lambda settings: SimpleNamespace(delete_branch=_record) + ) repo, pr_number, branch = "ClawHaven/acme-app", 93, "agent/eng-22" _park_work_item(repo=repo, pr_number=pr_number, branch=branch) @@ -280,7 +282,9 @@ async def test_external_close_deletes_branch_by_default(db_session, tmp_path, mo async def _record(repo, branch): deleted.append((repo, branch)) - monkeypatch.setattr(webhooks_mod, "_delete_branch", _record) + monkeypatch.setattr( + webhooks_mod, "get_github_client", lambda settings: SimpleNamespace(delete_branch=_record) + ) repo, pr_number, branch = "ClawHaven/acme-app", 94, "agent/eng-23" _park_work_item(repo=repo, pr_number=pr_number, branch=branch) @@ -310,7 +314,9 @@ async def _boom(cls, repo): async def _delete(repo, branch): deleted.append((repo, branch)) - monkeypatch.setattr(webhooks_mod, "_delete_branch", _delete) + monkeypatch.setattr( + webhooks_mod, "get_github_client", lambda settings: SimpleNamespace(delete_branch=_delete) + ) pushed = []