From 9a69458735f48887d860dd2148c6e5e00eecafd8 Mon Sep 17 00:00:00 2001 From: Paulo Date: Wed, 9 Sep 2026 19:57:06 +0200 Subject: [PATCH] Prove the Activity author contract with Field Notes (DRU-510) --- backend/druks/apps/base.py | 21 ---- .../druks_field_notes/contracts.py | 6 + .../druks_field_notes/subscribers.py | 5 +- .../druks_field_notes/workflows.py | 15 ++- .../druks-field_notes/tests/test_activity.py | 34 ++++++ backend/tests/test_durable_sdk.py | 84 ++++++++++++- backend/tests/test_proof_app.py | 6 +- backend/tests/test_proof_app_install.py | 6 +- docs/development.md | 6 +- docs/writing-an-app.md | 111 +++++++++++++----- frontend/src/lib/feed.test.ts | 11 +- frontend/src/lib/feed.ts | 3 +- 12 files changed, 248 insertions(+), 60 deletions(-) create mode 100644 backend/tests/druks-field_notes/tests/test_activity.py diff --git a/backend/druks/apps/base.py b/backend/druks/apps/base.py index 592fa652..6cf2bf44 100644 --- a/backend/druks/apps/base.py +++ b/backend/druks/apps/base.py @@ -8,7 +8,6 @@ from pydantic import BaseModel, Field, SecretStr -from druks.events.models import Event from druks.models import StoredSubject from druks.ui.exceptions import PageContractError, PageReadError, PageRouteError from druks.user_settings.models import SettingsOverride @@ -665,26 +664,6 @@ async def on_startup(cls) -> None: similar. The caller logs a failure and moves on, so one app can't wedge boot.""" - @classmethod - async def record_event( - cls, - *, - type: str, - subject: "Subject | StoredSubject | None" = None, - payload: dict[str, Any] | None = None, - ) -> None: - """Record one of this app's domain events to the log, stamped with the - app automatically. Apps record through here so the ``Event`` model - stays a platform internal. ``type`` is the milestone's own word ("merged") — - the feed reads it as one, so an app writes no rendering.""" - await Event.emit( - type=type, - subject=subject.identity if subject else None, - label=subject.label if subject else None, - payload=payload, - app=cls.name, - ) - @classmethod async def get_subject_progress( cls, subject: "Subject | StoredSubject" diff --git a/backend/tests/druks-field_notes/druks_field_notes/contracts.py b/backend/tests/druks-field_notes/druks_field_notes/contracts.py index 6d31dd2b..e51e0a59 100644 --- a/backend/tests/druks-field_notes/druks_field_notes/contracts.py +++ b/backend/tests/druks-field_notes/druks_field_notes/contracts.py @@ -4,3 +4,9 @@ class GistOutput(AgentOutput): # What the summarizer agent returns: the note it read, in one line. gist: str + + def get_artifact(self) -> dict[str, str]: + return {"kind": "markdown", "title": "Gist", "content": self.gist} + + def get_activity(self) -> dict[str, str]: + return {"kind": "gist.prepared", "summary": self.gist} diff --git a/backend/tests/druks-field_notes/druks_field_notes/subscribers.py b/backend/tests/druks-field_notes/druks_field_notes/subscribers.py index fe74bc2e..48d3efbe 100644 --- a/backend/tests/druks-field_notes/druks_field_notes/subscribers.py +++ b/backend/tests/druks-field_notes/druks_field_notes/subscribers.py @@ -1,13 +1,10 @@ from druks.signals import subscribe from druks.workflows import WorkflowEvent -from druks_field_notes.app import FieldNotes from druks_field_notes.models import Note from druks_field_notes.workflows import Summarize @subscribe(WorkflowEvent.FINISHED, workflow=Summarize) async def note_summarized(*, subject: Note, **_: object) -> None: - # A finished summarize is a milestone worth its own feed row. The workflow - # lifecycle is the trigger; the app only reacts. - await FieldNotes.record_event(type="summarized", subject=subject) + await subject.announce("note.gist_saved") diff --git a/backend/tests/druks-field_notes/druks_field_notes/workflows.py b/backend/tests/druks-field_notes/druks_field_notes/workflows.py index 316fad2a..f592f4f1 100644 --- a/backend/tests/druks-field_notes/druks_field_notes/workflows.py +++ b/backend/tests/druks-field_notes/druks_field_notes/workflows.py @@ -1,5 +1,5 @@ from druks.sandbox import Sandbox -from druks.workflows import Workflow +from druks.workflows import FatalError, Workflow from druks.workspaces import RepoWorkspace from druks_field_notes.app import FieldNotes @@ -41,3 +41,16 @@ async def run(self) -> None: @classmethod async def dispatch(cls, *, repository: Repository) -> str: return await cls.start(subject=repository) + + +class ApproveGist(Workflow): + """Prepare a gist and ask the operator to approve it.""" + + subject = Note + + async def run_multistep(self) -> None: + await FieldNotes.summarize(note_body=(await self.subject).body) + reply = await self.review() + if reply.action != "approve": + raise FatalError("The operator did not approve the gist.") + await self.announce("gist.approved") diff --git a/backend/tests/druks-field_notes/tests/test_activity.py b/backend/tests/druks-field_notes/tests/test_activity.py new file mode 100644 index 00000000..e55f7b46 --- /dev/null +++ b/backend/tests/druks-field_notes/tests/test_activity.py @@ -0,0 +1,34 @@ +from druks.apps import App +from druks.db import db_session +from druks.events import Event +from druks_field_notes.contracts import GistOutput +from druks_field_notes.models import Note, Repository +from druks_field_notes.subscribers import note_summarized +from sqlalchemy import select + + +def test_gist_declares_one_useful_result(): + output = GistOutput(gist="The pump ran hot.") + assert output.get_artifact() == { + "kind": "markdown", + "title": "Gist", + "content": "The pump ran hot.", + } + assert output.get_activity() == {"kind": "gist.prepared", "summary": "The pump ran hot."} + assert not hasattr(App, "record_event") + + +async def test_saved_note_and_repository_announce_distinct_domain_facts(druks_db): + db_session.registry.set(druks_db) + note = await Note.create(body="The pump ran hot.") + await note.save_gist("The pump ran hot.") + await note_summarized(subject=note) + repository = await Repository.create(repo="acme/pumps") + await repository.announce( + "repository.reported", summary="The source owner published its report." + ) + events = list(await druks_db.scalars(select(Event).order_by(Event.id))) + assert [event.type for event in events] == ["note.gist_saved", "repository.reported"] + assert {event.app for event in events} == {"field_notes"} + assert [event.subject_type for event in events] == ["note", "repository"] + assert all("run" not in event.payload for event in events) diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index 8ad0301d..c513a500 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -1,6 +1,7 @@ import asyncio import contextlib import os +from datetime import UTC, datetime from types import SimpleNamespace import psycopg @@ -13,12 +14,16 @@ from druks.durable import FatalError, Run, RunState from druks.durable.dbos_state import workflow_status from druks.durable.engine import configure_engine, init_dbos, launch, shutdown +from druks.durable.enums import AgentCallStatus from druks.durable.models import Artifact from druks.events.models import Event from druks.models import StoredSubject +from druks.sandbox.datastructures import AgentResult from druks.signals import subscribe from druks.testing import init_db -from druks.workflows import Gate, Subject, Workflow, step, task +from druks.workflows import Gate, OperatorReply, Subject, Workflow, step, task +from druks_field_notes.models import Note +from druks_field_notes.workflows import ApproveGist from pydantic import BaseModel from sqlalchemy import NullPool, create_engine, select from sqlalchemy.ext.asyncio import create_async_engine @@ -1436,3 +1441,80 @@ async def run_multistep(self) -> None: assert {event.payload["run"] for event in events} == {workflow_id} finally: workflows._items.pop(OutputFlow.kind) + + +async def test_field_notes_activity_through_admission_review_failure_and_replay( + rt, monkeypatch, tmp_path +): + monkeypatch.setenv("DRUKS_DATA_DIR", str(tmp_path)) + calls = [] + + @contextlib.asynccontextmanager + async def ephemeral(self, **kwargs): + async def run_agent(**kwargs): + calls.append(kwargs["call_id"]) + return AgentResult( + output={"gist": "The pump ran hot."}, + run_id=kwargs["call_id"], + sandbox_host_id="host-test", + model="claude", + agent=kwargs["agent"], + status=AgentCallStatus.SUCCEEDED, + started_at=datetime.now(UTC), + ) + + yield SimpleNamespace(run_agent=run_agent, id="host-test") + + monkeypatch.setattr("druks.sandbox.client.Client.ephemeral", ephemeral) + monkeypatch.setattr("druks.agents.render_prompt", _fake_render) + completed = [] + body = ApproveGist.run_multistep + + async def interrupt_after_approval(self): + await body(self) + completed.append(self.workflow_id) + if len(completed) == 1: + raise asyncio.CancelledError + + monkeypatch.setattr(ApproveGist, "run_multistep", interrupt_after_approval) + async with session_scope(rt.engine): + approved_note = await Note.create(body="The pump ran hot.") + rejected_note = await Note.create(body="The reading needs another pass.") + + approved_id = await ApproveGist.start(subject=approved_note) + await _wait_for(rt.engine, approved_id, lambda run: run.is_parked) + async with session_scope(rt.engine): + await OperatorReply.answer(approved_note, action="approve") + await _wait_for(rt.engine, approved_id, lambda run: len(completed) == 1) + await DBOS.resume_workflow_async(approved_id) + await _wait_for(rt.engine, approved_id, lambda run: run.state == RunState.FINISHED) + + rejected_id = await ApproveGist.start(subject=rejected_note) + await _wait_for(rt.engine, rejected_id, lambda run: run.is_parked) + async with session_scope(rt.engine): + await OperatorReply.answer(rejected_note, action="request_changes") + await _wait_for(rt.engine, rejected_id, lambda run: run.state == RunState.FAILED) + + assert completed == [approved_id, approved_id] + assert len(calls) == 2 + async with get_session(rt.engine) as session: + events = list( + await session.scalars(select(Event).filter_by(app="field_notes").order_by(Event.id)) + ) + for run_id, outcome in [(approved_id, "gist.approved"), (rejected_id, "workflow.failed")]: + history = [event for event in events if event.payload.get("run") == run_id] + for kind in ["workflow.scheduled", "gist.prepared", "workflow.parked", outcome]: + assert sum(event.type == kind for event in history) == 1 + request = next(event for event in history if event.type == "workflow.parked") + receipts = [ + event + for event in history + if event.type == "workflow.running" and "result" in event.payload + ] + assert len(receipts) == 1 + assert receipts[0].payload["gate"] == "review" + assert receipts[0].payload["input_requested_at"] == request.payload["input_requested_at"] + result = next(event for event in history if event.type == "gist.prepared") + assert result.payload["agent_call_id"] in calls + assert result.payload["artifact_id"] + assert sum(event.type == "gist.approved" for event in events) == 1 diff --git a/backend/tests/test_proof_app.py b/backend/tests/test_proof_app.py index fc91df14..f983e530 100644 --- a/backend/tests/test_proof_app.py +++ b/backend/tests/test_proof_app.py @@ -16,7 +16,11 @@ def test_discovery_registers_the_tables_and_capabilities(): app = load_app("field_notes") assert "field_notes_notes" in Base.metadata.tables - assert [workflow.__name__ for workflow in app.workflows()] == ["Summarize", "Survey"] + assert [workflow.__name__ for workflow in app.workflows()] == [ + "ApproveGist", + "Summarize", + "Survey", + ] capability_modules = {module.__name__ for module in app.capability_modules()} assert f"{_PACKAGE}.subscribers" in capability_modules diff --git a/backend/tests/test_proof_app_install.py b/backend/tests/test_proof_app_install.py index db2f40ba..4ffcc081 100644 --- a/backend/tests/test_proof_app_install.py +++ b/backend/tests/test_proof_app_install.py @@ -20,7 +20,11 @@ "sync_signing_key", "sync_token", ] - assert [workflow.__name__ for workflow in app.workflows()] == ["Summarize", "Survey"] + assert [workflow.__name__ for workflow in app.workflows()] == [ + "ApproveGist", + "Summarize", + "Survey", + ] assert {router.prefix for router in app.routers()} >= {"/notes", "/note"} assert app.migrations_dir() is not None print("ok") diff --git a/docs/development.md b/docs/development.md index d5ac8a8a..ae1e0606 100644 --- a/docs/development.md +++ b/docs/development.md @@ -97,7 +97,11 @@ runs the proof-app tests. Those tests are the executable contract for: - Role-module discovery - Route and subject read-side mounting - Independent migrations and table-prefix enforcement -- Workflow start, settings, and feed formatting. +- Workflow start, settings, and feed formatting +- Output artifacts and Activity declarations +- Accepted work, validated gate replies, and terminal failures +- Subject announcements on stored notes and repositories +- Completed-step replay without duplicate output or announcement rows. If you change the author API, update the scaffold, proof app, author guide, and tests together. diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index 5782e563..c1101b8f 100644 --- a/docs/writing-an-app.md +++ b/docs/writing-an-app.md @@ -10,6 +10,48 @@ Druks supplies durable execution and shared operating services. Read [the app boundary](concepts.md#the-app-boundary) before you assign ownership of a capability. +## Publish useful activity + +An agent output declares its saved result and its Activity kind: + +```python +from druks.agents import AgentOutput + + +class GistOutput(AgentOutput): + gist: str + + def get_artifact(self) -> dict[str, str]: + return {"kind": "markdown", "title": "Gist", "content": self.gist} + + def get_activity(self) -> dict[str, str]: + return {"kind": "gist.prepared", "summary": self.gist} +``` + +For a domain fact, use the subject you already hold: + +```python +await note.announce("note.gist_saved") +``` + +The independently installed Field Notes proof app uses these calls. Its +`ApproveGist` workflow uses `await self.review()` to request a decision. +An operator action answers through the public gate: + +```python +from druks.workflows import OperatorReply +from druks_field_notes.workflows import ApproveGist + +await ApproveGist.start(subject=note) +# After the run requests a decision: +await OperatorReply.answer(note, action="approve") +``` + +Druks records accepted work, the saved output, the request, and the validated +reply. The workflow announces its approval or raises a terminal failure. +Authors supply no routing IDs, timestamps, sessions, or event rows. +See [Activity facts](#activity-facts-and-signals) for the ownership rules. + ## Scaffold and prove the package ```bash @@ -402,7 +444,15 @@ refuses before any sandbox work when the login is missing, then provisions or attaches a sandbox, executes the CLI, validates the structured output, and records the call. Override `AgentOutput.to_result()` to map the strict agent contract to a domain value. -Override `get_artifact()` to publish a reviewable artifact. +Override `get_artifact()` to publish a reviewable artifact. Add `get_activity()` +to declare a non-empty `kind` and an optional string `summary`. An Activity +declaration requires an artifact. An invalid declaration fails the agent call +with `WorkflowError`. + +Druks records one Activity row with the artifact and its producing agent call. +The row and artifact share a transaction. Recovery reuses a completed call; +it does not create a second output row. Two new calls can produce two rows with +the same kind. The saved result has its own identity in each row. Pass `contract=OutputType` on an agent call when its required output fields depend on the input. Druks uses that type for the harness schema, validation, @@ -823,33 +873,38 @@ statuses = await Repository.get_statuses([summary.id for summary in summaries]) This is the read the platform's own board uses, so a declared page listing fifty rows costs one query rather than fifty. -## Record events and react to signals - -Record an event through the app. Druks stamps its ownership: - -```python -await NightWatch.record_event( - type="report.published", - subject=repository, - payload={"url": report_url}, -) -``` - -`type` is the milestone word that the feed reads. There is no presentation hook -to implement. Lifecycle events for subjected workflows are -recorded automatically. Call `record_event()` inside a platform-bound -transaction such as a request, durable step, or subscriber. - -A feed row contains facts, not prose. It contains its kind, workflow, subject -identity, and event payload. A client supplies the words. Give the subject a -``label`` for its one-line description. Each later event for the subject keeps -that label: - -```python -class Repository(StoredSubject): - def get_label(self) -> str: - return self.full_name -``` +## Activity facts and signals + +Use [announcements](#announcing-domain-events) for facts the app owns. +Use an agent output's `get_artifact()` and `get_activity()` for a saved result. +Do not announce that same result again from a completion subscriber. Field Notes +records `gist.prepared` for the artifact and `note.gist_saved` for the separate +change to its stored note. + +Druks records these workflow facts without app calls: + +- Accepted work: a new subject run entered the queue. A deduplicated start adds + no second row. +- Input requested: the workflow reached a gate. The record holds the gate and + the request time for that round. +- Response received: the concrete gate validated a reply. This does not mean + that the domain accepted the reply's proposal. The workflow decides that. +- Run failed: the run ended with a terminal failure. Routine running and + finished signals remain available to subscribers but do not enter Activity. +- Operator stopped: the operator stop route completed the stop. Domain cleanup + does not announce an owner outcome such as a closed pull request. + +An external owner can announce an outcome after the run stops. Record that +outcome when the owner reports it. Do not infer it from the run state. + +Each Activity row stores the subject label at the time of the event. Search +matches that recorded label. The client supplies readable labels from an app +catalog or the shared humanizer. For example, `gist.prepared` becomes +“Gist prepared.” Keep UI wording out of the event identity. + +A gate request and its reply retain the same run, gate, and request-time +identity. A result retains its artifact identity. These references describe the +recorded occurrence even when the current subject, run, or file is unavailable. React with filters rather than body guards: diff --git a/frontend/src/lib/feed.test.ts b/frontend/src/lib/feed.test.ts index 3303f58a..b390c23d 100644 --- a/frontend/src/lib/feed.test.ts +++ b/frontend/src/lib/feed.test.ts @@ -84,7 +84,7 @@ describe('eventLine', () => { }), ) - expect(line.label).toBe('summarized') + expect(line.label).toBe('Summarized') expect(line.subject).toBe('note 7') expect(line.path).toBeUndefined() }) @@ -100,3 +100,12 @@ it('retains the recorded run and decision round in Factory links', () => { expect(eventLine(event({ app: 'software_factory', subjectType: 'work_item', subjectId: '42', isSubjectAvailable: false })).path).toBeUndefined() }) + + +it.each([ + ['gist.prepared', 'Gist prepared'], + ['note.gist_saved', 'Note gist saved'], + ['gist.approved', 'Gist approved'], +])('gives %s readable words without an app formatter', (kind, label) => { + expect(eventLine(event({ kind, app: 'field_notes' })).label).toBe(label) +}) diff --git a/frontend/src/lib/feed.ts b/frontend/src/lib/feed.ts index 0ef7911d..826cd6e1 100644 --- a/frontend/src/lib/feed.ts +++ b/frontend/src/lib/feed.ts @@ -68,5 +68,6 @@ function localName(kind: string | null | undefined): string { } function words(identifier: string): string { - return identifier.replace(/_/g, ' ') + const text = identifier.replace(/[._]/g, ' ') + return text.charAt(0).toUpperCase() + text.slice(1) }