Prove the Activity author contract with Field Notes (DRU-510)

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"

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}

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

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

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)

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

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

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

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.

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: