diff --git a/backend/druks/agents.py b/backend/druks/agents.py index 18853db..18ee7e4 100644 --- a/backend/druks/agents.py +++ b/backend/druks/agents.py @@ -151,11 +151,17 @@ async def _do() -> Any: # a standalone run is its own memoized step + session async with step_session(): return await _invoke() - return await DBOS.run_step_async(StepOptions(name=f"{workflow.kind}.agent.{self.id}"), _do) + result = await DBOS.run_step_async( + StepOptions(name=f"{workflow.kind}.agent.{self.id}"), _do + ) + # 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) + return result async def _run(self, *, workflow_id: str, **context: Any) -> Any: """The raw execution: provision or attach a host, record the AgentCall, run - the harness. ``run_agent`` handles the durable wrapping + nesting.""" + the harness. ``__call__`` handles the durable wrapping + nesting.""" if not self.prompt: raise WorkflowError(f"agent {self.id!r} has no prompt template to render") model = self.get_model_name() diff --git a/backend/druks/build/workflows.py b/backend/druks/build/workflows.py index d44a2af..09bac95 100644 --- a/backend/druks/build/workflows.py +++ b/backend/druks/build/workflows.py @@ -279,12 +279,12 @@ async def _plan_phase(self) -> bool: ): return True reply = await self.review(questions=plan.questions) - if reply["action"] == "cancel": + if reply.action == "cancel": raise FatalError("cancelled at plan review") - if reply["action"] == "approve" and not plan.questions: + if reply.action == "approve" and not plan.questions: return True - answered = plan.get_answered(reply["answers"]) - note = reply["note"] + answered = plan.get_answered(reply.answers) + note = reply.note async def _implement_phase(self) -> None: while True: diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 0294493..6bc96bd 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -1,12 +1,12 @@ import inspect import re from collections.abc import Callable -from contextlib import nullcontext +from contextlib import nullcontext, suppress from contextvars import ContextVar from datetime import UTC, datetime from functools import partial from types import SimpleNamespace -from typing import TYPE_CHECKING, Any, ClassVar, Self, get_type_hints +from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self, TypeVar, get_args, get_type_hints from croniter import croniter from dbos import DBOS, SetEnqueueOptions, SetWorkflowAttributes, SetWorkflowID, StepOptions @@ -50,7 +50,9 @@ "AgentCallStatus", "FatalError", "Gate", + "ReviewReply", "Run", + "RunRecord", "RunState", "SubjectActivity", "SubjectSummary", @@ -67,13 +69,13 @@ # A human gate can park for days; a long recv TTL still caps zombie parks. GATE_TTL_SECONDS = 14 * 24 * 60 * 60 -# The controls an in-app review always offers. Framework-owned: an extension's agent -# supplies the plan and questions (content), but the decision verbs are ours — so a +# The decision verbs an in-app review offers. Framework-owned: an extension's +# agent supplies the plan and questions (content), but the verbs are ours — so a # resume can only carry an action we defined, the line the resume endpoint checks. -_REVIEW_CONTROLS = ("approve", "request_changes", "cancel") -# The recv topic every in-app review parks on. A run parks on one gate at a time, so -# one topic serves them all; Run.resume routes the reply back through it. -_REVIEW_TOPIC = "review" +_ReviewAction = Literal["approve", "request_changes", "cancel"] +_REVIEW_CONTROLS = get_args(_ReviewAction) + +T = TypeVar("T") # The running workflow instance, so a Gate's on_wait() can reach its extension's # side-effects (set draft, request review, …) when the gate parks. No default: @@ -163,6 +165,31 @@ 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``.""" + + def __init__(self) -> None: + self._entries: list[Any] = [] + + def _append(self, entry: Any) -> None: + self._entries.append(entry) + + def list(self, contract: type[T], **filters: Any) -> list[T]: + return [ + entry + for entry in self._entries + if isinstance(entry, contract) + and all(getattr(entry, name) == expected for name, expected in filters.items()) + ] + + def latest(self, contract: type[T], **filters: Any) -> T | None: + with suppress(IndexError): + return self.list(contract, **filters)[-1] + return + + class Gate(BaseModel): """A typed human-in-the-loop gate. Subclass per park point; the class name is the durable recv topic, the fields are the reply's schema. `wait()` parks the @@ -207,7 +234,20 @@ 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) - return cls.model_validate(payload) + reply = cls.model_validate(payload) + workflow.record._append(reply) + return reply + + +class ReviewReply(Gate): + """The operator's in-app review decision. Answers and note are content for + the next agent prompt, never control flow.""" + + # One topic serves every in-app review — a run parks on one gate at a time. + topic = "review" + action: _ReviewAction + answers: dict[str, str] = Field(default_factory=dict) + note: str = "" async def _park( @@ -477,6 +517,9 @@ 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() # 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] = {} @@ -513,24 +556,26 @@ 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) -> dict[str, Any]: - # Park for an in-app decision: the ask carries the controls and any questions; - # the reply is {action, answers, note} — an answer is an offered option id or - # the operator's own words, note their free-text remark. Both are content for - # the next agent prompt, never control flow (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 demand. + 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 + # 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 + # demand. 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(_REVIEW_TOPIC) + raise SubjectlessGate(ReviewReply.topic) request = { "presentation": "in_app", "controls": list(_REVIEW_CONTROLS), "questions": [q.model_dump(mode="json") for q in questions or ()], } - return await _park(self, _REVIEW_TOPIC, request, GATE_TTL_SECONDS) + payload = await _park(self, ReviewReply.topic, request, GATE_TTL_SECONDS) + reply = ReviewReply.model_validate(payload) + self.record._append(reply) + return reply async def get_prompt_context(self, **context: Any) -> dict[str, Any]: # Everything an agent's template renders with, beyond the workflow and @@ -635,7 +680,7 @@ async def start( # fails at start, not inside the run. # subject is required (no default) so a run can't silently lose its # timeline by omission — pass subject=None explicitly for a background run. - if account_id is None: + if not account_id: # Browser-origin starts inherit the request's authenticated account; # dispatchers that know better pass account_id explicitly. account_id = current_account_id.get() diff --git a/backend/tests/test_build_plan_phase.py b/backend/tests/test_build_plan_phase.py index c96b294..af75005 100644 --- a/backend/tests/test_build_plan_phase.py +++ b/backend/tests/test_build_plan_phase.py @@ -5,7 +5,7 @@ from druks.build.enums import ReviewDecision from druks.build.policy import RepoPolicy from druks.build.workflows import Build, BuildWorkflow -from druks.workflows import FatalError +from druks.workflows import FatalError, ReviewReply async def test_plan_phase_threads_free_text_into_the_next_pass(monkeypatch): @@ -48,13 +48,11 @@ async def fake_review_agent(): replies = iter( [ - { - "action": "request_changes", - "answers": {"q1": "memcache — redis is banned here"}, - "note": "", - }, - {"action": "request_changes", "answers": {}, "note": "add a rollback section"}, - {"action": "approve", "answers": {}, "note": ""}, + ReviewReply( + action="request_changes", answers={"q1": "memcache — redis is banned here"} + ), + ReviewReply(action="request_changes", note="add a rollback section"), + ReviewReply(action="approve"), ] ) diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index 7e45bec..ea54a54 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -67,11 +67,11 @@ class Confirm(Gate): class SampleFlow(Workflow): @step - async def record(self, repo: str) -> str: + async def note_repo(self, repo: str) -> str: return f"recorded:{repo}" async def run_multistep(self, repo: str) -> None: - await self.record(repo) + await self.note_repo(repo) decision = await Approve.wait() if decision.action == "close": raise FatalError("closed at review") @@ -81,16 +81,27 @@ 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))}") 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. + 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}") + class RecordFeedback(Workflow): @step - async def record(self, repo: str) -> None: + async def note_repo(self, repo: str) -> None: SINK.append(repo) async def run_multistep(self, repo: str) -> None: - await self.record(repo) + await self.note_repo(repo) # every= so launch()'s apply_schedules has a schedule to create (smoke). class DailySweep(Workflow): @@ -114,6 +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}") class ConfirmFlow(Workflow): async def run_multistep(self) -> None: @@ -135,6 +148,7 @@ async def run_multistep(self) -> None: return ( SampleFlow, AgentFlow, + AgentBodyFlow, RecordFeedback, SubjectFlow, DoubleGateFlow, @@ -187,6 +201,7 @@ def rt(): ( sample_flow, agent_flow, + agent_body_flow, feedback_flow, subject_flow, double_gate_flow, @@ -202,6 +217,7 @@ def rt(): engine=engine, SampleFlow=sample_flow, AgentFlow=agent_flow, + AgentBodyFlow=agent_body_flow, RecordFeedback=feedback_flow, SubjectFlow=subject_flow, DoubleGateFlow=double_gate_flow, @@ -217,6 +233,7 @@ def rt(): agents._items.pop("decider", None) workflows._items.pop("sample_flow", None) workflows._items.pop("agent_flow", None) + workflows._items.pop("agent_body_flow", None) workflows._items.pop("record_feedback", None) workflows._items.pop("daily_sweep", None) workflows._items.pop("subject_flow", None) @@ -373,6 +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 async def test_fail_branch(rt): @@ -433,17 +452,14 @@ async def test_subjectless_review_fails_loudly(rt): assert "'review'" in failed.failure -async def test_run_agent_step(rt, monkeypatch): - # Stub the VM; assert the step records an AgentCall and the result round-trips. +def _fake_ephemeral_returning(action: str, seen: list[dict], pinned: list[int]): + # Class-method stand-in for Client.ephemeral: a VM whose agent returns + # ``action``. from datetime import UTC, datetime from druks.durable.enums import AgentCallStatus - from druks.durable.models import AgentCall from druks.sandbox.datastructures import AgentResult - seen: list[dict] = [] - pinned: list[int] = [] - @contextlib.asynccontextmanager async def _fake_ephemeral(self, **_kw): async def _run_agent(**kwargs): @@ -457,7 +473,7 @@ async def _run_agent(**kwargs): # The harness names the on-disk dir (and the row) from the supplied # call_id, so the result echoes it back as run_id. return AgentResult( - output={"action": "stop"}, + output={"action": action}, run_id=kwargs["call_id"], sandbox_host_id="host-test", model="claude", @@ -469,17 +485,30 @@ async def _run_agent(**kwargs): # The base Workspace wrapping this box reads host_id off ``id``. yield SimpleNamespace(run_agent=_run_agent, id="host-test") - async def _fake_render(*_a, **_k): - return "PROMPT" + return _fake_ephemeral + + +async def _fake_render(*_a, **_k): + return "PROMPT" + + +async def test_run_agent_step(rt, monkeypatch): + # Stub the VM; assert the step records an AgentCall and the result round-trips. + from druks.durable.models import AgentCall + seen: list[dict] = [] + pinned: list[int] = [] # Patch the class method (not the singleton instance): an instance-attr # patch leaves a shadowing leftover that breaks later sandbox tests. - monkeypatch.setattr("druks.sandbox.client.Client.ephemeral", _fake_ephemeral) + monkeypatch.setattr( + "druks.sandbox.client.Client.ephemeral", _fake_ephemeral_returning("stop", seen, pinned) + ) monkeypatch.setattr("druks.agents.render_prompt", _fake_render) 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 seen and seen[0]["artifact_dir"].name == f"run-{wfid}" assert seen[0]["agent"] == "decider" session = get_session(rt.engine) @@ -495,6 +524,22 @@ async def _fake_render(*_a, **_k): 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.""" + seen: list[dict] = [] + pinned: list[int] = [] + monkeypatch.setattr( + "druks.sandbox.client.Client.ephemeral", _fake_ephemeral_returning("ship", seen, pinned) + ) + monkeypatch.setattr("druks.agents.render_prompt", _fake_render) + + 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 + + async def test_task_enqueue(rt): SINK.clear() await rt.RecordFeedback.start(subject=None, repo="owner/queued") diff --git a/backend/tests/test_run_record.py b/backend/tests/test_run_record.py new file mode 100644 index 0000000..979ad6b --- /dev/null +++ b/backend/tests/test_run_record.py @@ -0,0 +1,60 @@ +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 5005093..3d533c7 100644 --- a/docs/writing-an-extension.md +++ b/docs/writing-an-extension.md @@ -146,36 +146,41 @@ 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. -### Replay-rebuilt workflow state +### The run record -A `run_multistep()` body often accumulates working state across its calls — the -drafts produced so far, the reviews of the current draft, the last delivery. -Hold that state in plain instance attributes and mutate them only from -orchestration-body code, right after the memoized call that produced the value: +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: ```python -class Sweep(Workflow): - def __init__(self) -> None: - super().__init__() - self._findings: list[str] = [] - - async def run_multistep(self, repo: str) -> None: - self._findings.extend(await self.scan(repo)) +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 ``` -This is durable by determinism: recovery re-runs the body from the top with -every `@step`, agent call, and gate reply memoized, so the attributes rebuild -to exactly what the live pass held. It is the standard durable-execution idiom — -Temporal workflows hold state the same way. - -The one trap: never mutate these attributes inside a `@step` or `run()`. A -completed step is skipped on replay — its body never runs again — so a write -made there silently vanishes from the rebuilt state. A step returns values; -the replayed body records them. - -`BuildJournal` (`backend/druks/build/journal.py`) is the reference shape: one -typed object owning the run's working memory, with this contract stated on the -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. ### Schedules and settings