diff --git a/backend/druks/api/runs.py b/backend/druks/api/runs.py index c4b4ef22..5524b62d 100644 --- a/backend/druks/api/runs.py +++ b/backend/druks/api/runs.py @@ -10,8 +10,10 @@ agent_error_responses, ) from druks.api.schemas import CancelRunResponse, ResumeRequest, RetryRunResponse -from druks.durable.enums import RunState +from druks.apps.registry import workflows +from druks.durable.enums import RunState, WorkflowEvent from druks.durable.models import Run +from druks.events.models import Event from druks.notifications.exceptions import InvalidChoiceError from druks.notifications.services import validate_in_app_answer @@ -72,7 +74,16 @@ async def cancel_run( return CancelRunResponse(run=run.id, result="already_cancelled") if not run.is_active: raise RunNotActive(run_id) + subject = await run.get_subject() + label = run.subject_label await run.cancel(failure=reason) + await Event.emit( + type=WorkflowEvent.CANCELLED, + subject=subject, + label=label, + payload={"run": run.id, "kind": run.kind, "reason": reason}, + app=workflows.get(run.kind).app, + ) return CancelRunResponse(run=run.id, result="cancelled") diff --git a/backend/druks/apps/base.py b/backend/druks/apps/base.py index 4c29a05a..592fa652 100644 --- a/backend/druks/apps/base.py +++ b/backend/druks/apps/base.py @@ -31,7 +31,7 @@ from druks.agents import Agent from druks.doctor import CheckResult from druks.durable.datastructures import Subject - from druks.durable.schemas import SubjectActivity + from druks.durable.schemas import SubjectProgress from druks.ui.page import PageRoute from druks.workflows import Workflow @@ -574,7 +574,7 @@ async def upload(file: UploadFile) -> FileSummary: def _get_subject_routes( cls, subject_class: "type[Subject] | type[StoredSubject]" ) -> "APIRouter": - """The board and one subject (header + status + timeline + activity), each with a + """The board and one subject (header + status + timeline + progress), each with a point-in-time read and a ``/stream`` that pushes the whole snapshot on change. Mounted at ``/api//`` for every subject the app's workflows declare. Every read here is keyed by identity, so an app that @@ -613,7 +613,7 @@ async def subject_response(subject_id: str) -> SubjectResponse | None: subject_type, subject_id, summary=subject.get_summary(), - activity=await cls.get_subject_activity(subject), + progress=await cls.get_subject_progress(subject), ) @router.get("", response_model=SubjectList, response_model_by_alias=True) @@ -686,9 +686,8 @@ async def record_event( ) @classmethod - async def get_subject_activity( + async def get_subject_progress( cls, subject: "Subject | StoredSubject" - ) -> "SubjectActivity | None": - """The subject's live sub-phase, if any (e.g. "Provisioning sandbox VM…"). Optional — - override to surface a transient signal the running run pushes.""" + ) -> "SubjectProgress | None": + """Return labeled live detail from the current run's transient phase.""" return diff --git a/backend/druks/contrib/software_factory/app.py b/backend/druks/contrib/software_factory/app.py index 9b1f8765..ce9abfd3 100644 --- a/backend/druks/contrib/software_factory/app.py +++ b/backend/druks/contrib/software_factory/app.py @@ -21,15 +21,15 @@ from druks.db import StoredSubject from druks.doctor import CheckResult from druks.services import ServiceNotConnectedError -from druks.workflows import SubjectActivity +from druks.workflows import SubjectProgress from .services import GithubReviewer # Only what the timeline can't already show. A running agent has an agent call # to name it, so the phase that clears provisioning maps to nothing. -_PHASE_META: dict[str, SubjectActivity] = { - "provisioning_vm": SubjectActivity(label="Provisioning sandbox VM…", kind="infra"), - "sandbox_building": SubjectActivity(label="Building sandbox…", kind="infra"), +_PHASE_META: dict[str, SubjectProgress] = { + "provisioning_vm": SubjectProgress(label="Provisioning sandbox VM…", kind="infra"), + "sandbox_building": SubjectProgress(label="Building sandbox…", kind="infra"), } @@ -198,6 +198,6 @@ async def get_tracker(cls, source: str | None = None) -> Tracker | None: ) @classmethod - async def get_subject_activity(cls, subject: StoredSubject) -> SubjectActivity | None: + async def get_subject_progress(cls, subject: StoredSubject) -> SubjectProgress | None: phase = await subject.get_phase() return _PHASE_META.get(phase or "") diff --git a/backend/druks/durable/__init__.py b/backend/druks/durable/__init__.py index e9f91a8c..73d7ccee 100644 --- a/backend/druks/durable/__init__.py +++ b/backend/druks/durable/__init__.py @@ -2,7 +2,7 @@ from .enums import AgentCallStatus, RunState from .exceptions import FatalError, WorkflowError from .models import AgentCall, Run -from .schemas import AgentCallResponse, SubjectActivity, SubjectSummary +from .schemas import AgentCallResponse, SubjectProgress, SubjectSummary # The durable-execution engine. Internal — authors never import druks.durable; the # doors are druks.workflows (Workflow, Gate, step + these records) and druks.agents @@ -15,7 +15,7 @@ "FatalError", "Run", "RunState", - "SubjectActivity", + "SubjectProgress", "SubjectSummary", "WorkflowError", "get_run_phase", diff --git a/backend/druks/durable/datastructures.py b/backend/druks/durable/datastructures.py index dc84d9e6..ebb4b7ac 100644 --- a/backend/druks/durable/datastructures.py +++ b/backend/druks/durable/datastructures.py @@ -38,6 +38,22 @@ def label(self) -> str: # the handle, not a surrogate key. return self.id + async def announce(self, topic: str, **facts: Any) -> None: + """Record and deliver a domain fact in the current transaction.""" + # The app loader imports the durable package during registration. + from druks.apps.loader import resolve_workflow_app + from druks.events.models import Event + from druks.signals import publish + + await Event.emit( + type=topic, + subject=self.identity, + label=self.label, + payload=facts, + app=resolve_workflow_app(type(self).__module__), + ) + await publish(topic, subject=self.identity, **facts) + @classmethod async def get_for_subject_id(cls, subject_id: str) -> Self | None: """The subject this id names. Ids reach the read side as free text off a URL, diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py index ef584e67..ed4bb560 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -21,7 +21,7 @@ ArtifactDescriptor, ArtifactFile, RunResponse, - SubjectActivity, + SubjectProgress, SubjectResponse, SubjectStatus, SubjectSummary, @@ -104,7 +104,7 @@ async def get_subject_response( subject_id: str, *, summary: SubjectSummary, - activity: SubjectActivity | None = None, + progress: SubjectProgress | None = None, ) -> SubjectResponse: # list_for_subject is newest-first, so runs[0] is the driving run the status # reads — the same row get_latest_for_subject would return, its calls already @@ -115,7 +115,7 @@ async def get_subject_response( summary=summary, status=await _status(latest), timeline=await _timeline(runs), - activity=activity, + progress=progress, ) diff --git a/backend/druks/durable/schemas.py b/backend/druks/durable/schemas.py index 646718f2..e6167d8b 100644 --- a/backend/druks/durable/schemas.py +++ b/backend/druks/durable/schemas.py @@ -196,9 +196,9 @@ class SubjectList(Schema): rows: list[SubjectRow] = Field(default_factory=list) -class SubjectActivity(Schema): - # The running sub-phase the timeline can't show ("Provisioning sandbox VM…"), supplied - # by the app; ``kind`` groups it for display ("infra" | "agent"). +class SubjectProgress(Schema): + """The app's labeled live detail, grouped by kind for display.""" + label: str kind: str @@ -208,7 +208,7 @@ class SubjectResponse(Schema): status: SubjectStatus # The subject's runs, oldest first, each with its agent calls — the timeline. timeline: list[RunResponse] = Field(default_factory=list) - activity: SubjectActivity | None = None + progress: SubjectProgress | None = None class TranscriptChunk(Schema): diff --git a/backend/druks/events/models.py b/backend/druks/events/models.py index bf6baff6..c924bf81 100644 --- a/backend/druks/events/models.py +++ b/backend/druks/events/models.py @@ -3,6 +3,7 @@ from sqlalchemy import Index from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, mapped_column from druks.database import db_session @@ -10,9 +11,7 @@ class Event(Base): - """The append-only log: one row per run-state transition and (later) domain - milestone, keyed to the subject it concerns. An app reads it back as a feed, - or folds the newest-per-subject into a status.""" + """Recorded workflow and domain facts, keyed to their subject.""" __tablename__ = "events" # Newest-per-subject is the history/dashboard rollup; the feed orders on the @@ -42,9 +41,12 @@ async def emit( label: str | None = None, payload: dict[str, Any] | None = None, app: str | None = None, + session: AsyncSession | None = None, ) -> None: + """Record in the supplied session or the current domain transaction.""" + session = session or db_session() subject = subject or {} - db_session().add( + session.add( cls( type=type, subject_type=subject.get("type"), @@ -54,4 +56,4 @@ async def emit( payload=payload or {}, ) ) - await db_session().flush() + await session.flush() diff --git a/backend/druks/models.py b/backend/druks/models.py index b7e38640..152d310f 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -74,6 +74,22 @@ def get_label(self) -> str: def label(self) -> str: return self.get_label() + async def announce(self, topic: str, **facts: Any) -> None: + """Record and deliver a domain fact in the current transaction.""" + # The loader and event model depend on this module's Base. + from druks.apps.loader import resolve_workflow_app + from druks.events.models import Event + from druks.signals import publish + + await Event.emit( + type=topic, + subject=self.identity, + label=self.label, + payload=facts, + app=resolve_workflow_app(type(self).__module__), + ) + await publish(topic, subject=self.identity, **facts) + @classmethod async def get_for_subject_id(cls, subject_id: str) -> Self | None: """The row this subject id names. A subject id is free text and reaches the diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index cd680433..a4c5a605 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -35,6 +35,7 @@ validate_setting_override, validate_settings_declaration, ) +from druks.database import get_session from druks.durable.activity import set_run_phase from druks.durable.datastructures import Subject from druks.durable.engine import _step_engine, register_schedule, run_queue, step_session @@ -44,7 +45,7 @@ from druks.durable.schemas import ( AgentCallResponse, RunResponse, - SubjectActivity, + SubjectProgress, SubjectStatus, SubjectSummary, ) @@ -74,7 +75,7 @@ "OperatorReply", "RunResponse", "Subject", - "SubjectActivity", + "SubjectProgress", "SubjectStatus", "SubjectSummary", "Workflow", @@ -99,6 +100,7 @@ _REVIEW_CONTROLS = get_args(_ReviewAction) T = TypeVar("T") +GateReply = TypeVar("GateReply", bound="Gate") # The running workflow instance, so a Gate's on_wait() can reach its app's # side-effects (set draft, request review, …) when the gate parks. No default: @@ -315,8 +317,7 @@ async def _on_wait() -> None: await cls.on_wait(workflow) await DBOS.run_step_async(StepOptions(name=f"{cls.name}._on_wait"), _on_wait) - payload = await _park(workflow, cls.name, input_request, ttl_seconds) - reply = cls.model_validate(payload) + reply = await _park(workflow, cls, input_request, ttl_seconds) workflow.journal.add(reply) return reply @@ -334,10 +335,10 @@ class OperatorReply(Gate): async def _park( workflow: "Workflow", - gate: str, + gate: type[GateReply], input_request: dict[str, Any] | None, ttl_seconds: float, -) -> dict[str, Any]: +) -> GateReply: # Shared park core: a park lasts days, so reap the warm VM, then suspend on the # gate's channel until Run.resume answers it. await workflow._reap_run() @@ -346,7 +347,7 @@ async def _park( RunState.PARKED, subject=workflow._subject, facts={ - "input_gate": gate, + "input_gate": gate.name, "input_request": input_request, "input_requested_at": datetime.now(UTC), }, @@ -354,22 +355,22 @@ async def _park( if workflow._subject: # Every subjected park notifies the designated destination — no author opt-in. await _notify_designated_destination(workflow.workflow_id, workflow._subject) - payload = await DBOS.recv_async(gate, timeout_seconds=ttl_seconds) + payload = await DBOS.recv_async(gate.name, timeout_seconds=ttl_seconds) if payload is None: - raise GateTimeout(gate) + raise GateTimeout(gate.name) + reply = gate.model_validate(payload) await _emit_run_event( workflow.workflow_id, RunState.RUNNING, subject=workflow._subject, facts={**_GATE_CLEARED, "answer_parked_at": Run.input_requested_at}, + result=reply, ) - return payload + return reply async def _notify_designated_destination(workflow_id: str, subject: dict[str, Any]) -> None: - # Reads the ask off the run row the parked step just wrote (the - # signal payload carries no ask — producer-side placement is the point); - # the settings pointer is the operator's off-switch. + # The operator's settings select the destination for the recorded request. async def _create() -> str | None: async with step_session(): run = await Run.get(workflow_id) @@ -553,6 +554,7 @@ async def _transition() -> dict[str, Any] | None: # Read before the flush: flushing the update unloads the row's # computed columns, and reading one back would be implicit IO. label = run.subject_label + gate = run.input_gate if state == RunState.RUNNING and result else None if facts: for field, value in facts.items(): setattr(run, field, value) @@ -562,7 +564,7 @@ async def _transition() -> dict[str, Any] | None: return { "kind": run.kind, "subject": subject, - "payload": await _log_run_event(run, state, subject, label, result), + "payload": await _log_run_event(run, state, subject, label, result, gate), } transition = await DBOS.run_step_async( @@ -591,14 +593,19 @@ async def _log_run_event( subject: dict[str, Any], label: str | None, result: Any = None, + gate: str | None = None, ) -> dict[str, Any]: # One event per transition — the feed's run-level granularity, read off the # just-written row so gate and failure ride the transition that set them. # The result rides the finished event so reactions read the outcome off the # payload instead of artifacts. payload: dict[str, Any] = {"run": run.id, "kind": run.kind} - if run.input_gate: - payload["gate"] = run.input_gate + if gate or run.input_gate: + payload["gate"] = gate or run.input_gate + if gate or state == RunState.PARKED: + payload["input_requested_at"] = run.input_requested_at.isoformat() + if state == RunState.PARKED: + payload["input_request"] = run.input_request if run.failure: payload["failure"] = run.failure if isinstance(result, BaseModel): @@ -779,19 +786,28 @@ def __init__(self) -> None: self._host_secrets_id = "" async def announce(self, topic: str, **facts: Any) -> None: - # The workflow announcing a domain event in its app's vocabulary - # ("pr.opened", pr_number=12, branch="agent/eng-8"). The platform injects - # the routing subscribers filter on, and the publish runs as its own - # retrying checkpoint so a recovery replay doesn't re-fire it. Body-only, - # enforced: the checkpoint is a step, so it can't nest inside one. + """Record a domain fact, then notify subscribers in a separate checkpoint.""" if _in_step.get(): raise WorkflowError("announce() runs in the workflow body, not inside a @step") - async def _fan_out() -> None: + async def record() -> None: + async with step_session(): + run = await Run.get(self.workflow_id) + await Event.emit( + type=topic, + subject=self._subject, + label=run.subject_label, + payload={**facts, "run": self.workflow_id, "kind": self.kind}, + app=self.app, + ) + + await DBOS.run_step_async(StepOptions(name=topic, **_IO_RETRIES), record) + + async def notify() -> None: async with step_session(): await publish(topic, subject=self._subject, kind=self.kind, **facts) - await DBOS.run_step_async(StepOptions(name=topic, **_IO_RETRIES), _fan_out) + await DBOS.run_step_async(StepOptions(name=f"{topic}:propagate", **_IO_RETRIES), notify) async def review( self, *, questions: list[BaseModel] | None = None, context: str = "" @@ -814,8 +830,7 @@ async def review( } if context: request["context"] = context - payload = await _park(self, OperatorReply.name, request, GATE_TTL_SECONDS) - reply = OperatorReply.model_validate(payload) + reply = await _park(self, OperatorReply, request, GATE_TTL_SECONDS) self.journal.add(reply) return reply @@ -1030,6 +1045,18 @@ async def start( _step_engine(), workflow_id=workflow_id, kind=cls.kind, account_id=account_id ) if subject: + if cls.app: + # Admission must be visible before the caller's transaction commits. + async with get_session(_step_engine()) as session: + await Event.emit( + type=WorkflowEvent.SCHEDULED, + subject=subject_record, + label=subject.label, + payload={"run": workflow_id, "kind": cls.kind}, + app=cls.app, + session=session, + ) + await session.commit() await publish(WorkflowEvent.SCHEDULED, subject=subject.identity, kind=cls.kind) return workflow_id # The slot was held — the handle is the subject's live run. diff --git a/backend/tests/software_factory/test_api_work_items.py b/backend/tests/software_factory/test_api_work_items.py index 45038c29..c54d85ab 100644 --- a/backend/tests/software_factory/test_api_work_items.py +++ b/backend/tests/software_factory/test_api_work_items.py @@ -2,9 +2,10 @@ from pathlib import Path import pytest +from druks.contrib.software_factory.app import SoftwareFactory from druks.contrib.software_factory.models import WorkItem -from druks.durable.reads import list_subject_timeline -from druks.testing import asgi_client, seed_call +from druks.durable.reads import get_subject_phase, list_subject_timeline +from druks.testing import asgi_client, configure_app_for_test, make_settings, seed_call from fastapi.testclient import TestClient from software_factory.factories import make_test_work_item, seed_build_run @@ -17,16 +18,11 @@ } -def _build_app(tmp_path): - from druks.testing import configure_app_for_test, make_settings - - settings = make_settings(tmp_path) - return configure_app_for_test(settings=settings) - - @pytest.fixture async def client(tmp_path: Path, druks_db): - async with asgi_client(_build_app(tmp_path)) as client: + app = configure_app_for_test(settings=make_settings(tmp_path)) + + async with asgi_client(app) as client: yield client @@ -244,11 +240,7 @@ async def test_timeline_shows_every_build_attempt(druks_db): assert any(e.failure == "boom" for e in entries) -async def test_subject_activity_surfaces_running_phase(druks_db, monkeypatch): - # A running build run pushes a transient phase; the detail view's live activity - # surfaces it ("Provisioning sandbox VM…") — finer than the lifecycle status. - from druks.contrib.software_factory import app as software_factory_app - +async def test_subject_progress_surfaces_running_phase(client, druks_db, monkeypatch): item = await make_test_work_item(repo="ClawHaven/acme-app", title="x") await seed_build_run(druks_db, work_item_id=item.id, state="running") @@ -256,17 +248,24 @@ async def phase(_run_id): return "provisioning_vm" monkeypatch.setattr("druks.durable.reads.get_run_phase", phase) - activity = await software_factory_app.SoftwareFactory.get_subject_activity(item) - assert activity is not None - assert activity.label == "Provisioning sandbox VM…" - assert activity.kind == "infra" - + progress = await SoftwareFactory.get_subject_progress(item) + assert progress + assert progress.label == "Provisioning sandbox VM…" + assert progress.kind == "infra" + assert await item.get_phase() == "provisioning_vm" + assert await get_subject_phase(item.subject_type, str(item.id)) == "provisioning_vm" + + response = await client.get(f"/api/software_factory/work_item/{item.id}") + assert response.status_code == 200 + assert response.json()["progress"] == { + "label": "Provisioning sandbox VM…", + "kind": "infra", + } + assert "activity" not in response.json() -async def test_subject_activity_none_when_not_running(druks_db): - # A run parked on a gate isn't working — no live sub-phase. - from druks.contrib.software_factory import app as software_factory_app +async def test_subject_progress_none_when_not_running(druks_db): item = await make_test_work_item(repo="ClawHaven/acme-app", title="x") await seed_build_run(druks_db, work_item_id=item.id, state="parked", input_gate="review_plan") - assert await software_factory_app.SoftwareFactory.get_subject_activity(item) is None + assert await SoftwareFactory.get_subject_progress(item) is None diff --git a/backend/tests/test_announcements.py b/backend/tests/test_announcements.py new file mode 100644 index 00000000..0930dc23 --- /dev/null +++ b/backend/tests/test_announcements.py @@ -0,0 +1,80 @@ +import pytest +from druks.apps import loader +from druks.db import db_session +from druks.events.models import Event +from druks.signals import subscribe +from druks.workflows import Subject +from druks_field_notes.models import Note +from sqlalchemy import select + + +class Report(Subject): + pass + + +async def test_identity_subject_uses_its_registered_app(druks_db, monkeypatch): + monkeypatch.setitem(loader._workflow_packages, __name__, "reports") + report = Report(id="owner/repository#7") + received = [] + + @subscribe("report.published", subject=Report) + async def receive(*, subject: Report, url: str) -> None: + received.append((subject, url)) + + await report.announce("report.published", url="https://example.com/report/7") + + event = (await druks_db.scalars(select(Event).filter_by(type="report.published"))).one() + assert event.app == "reports" + assert event.subject_type == "report" + assert event.subject_id == report.id + assert event.subject_label == report.label + assert event.payload == {"url": "https://example.com/report/7"} + assert received == [(report, "https://example.com/report/7")] + + +async def test_stored_subject_keeps_distinct_domain_changes(druks_db): + note = await Note.create(body="An observation") + + await note.announce("note.revised", revision=1) + await note.announce("note.revised", revision=2) + + events = list( + await druks_db.scalars(select(Event).filter_by(type="note.revised").order_by(Event.id)) + ) + assert [event.payload for event in events] == [{"revision": 1}, {"revision": 2}] + assert {event.app for event in events} == {"field_notes"} + assert {event.subject_id for event in events} == {str(note.id)} + assert {event.subject_label for event in events} == {note.label} + + +async def test_domain_rollback_removes_the_change_and_announcement(druks_db): + db_session.registry.set(druks_db) + note = await Note.create(body="An observation") + + with pytest.raises(ValueError, match="Rejected change"): + async with druks_db.begin_nested(): + note.gist = "A summary" + await note.announce("note.summarized", gist=note.gist) + raise ValueError("Rejected change") + + await druks_db.refresh(note) + assert note.gist is None + assert not list(await druks_db.scalars(select(Event).filter_by(type="note.summarized"))) + + +async def test_subject_delivery_error_rolls_back_with_the_domain_transaction(druks_db): + db_session.registry.set(druks_db) + note = await Note.create(body="An observation") + + @subscribe("note.delivery_failed", subject=Note) + async def fail(**facts: object) -> None: + raise RuntimeError("Subscriber unavailable") + + with pytest.raises(RuntimeError, match="Subscriber unavailable"): + async with druks_db.begin_nested(): + note.gist = "A summary" + await note.announce("note.delivery_failed") + + await druks_db.refresh(note) + assert note.gist is None + assert not list(await druks_db.scalars(select(Event).filter_by(type="note.delivery_failed"))) diff --git a/backend/tests/test_author_surface.py b/backend/tests/test_author_surface.py index 5f9dc5de..3ec798e2 100644 --- a/backend/tests/test_author_surface.py +++ b/backend/tests/test_author_surface.py @@ -27,7 +27,7 @@ "OperatorReply", "RunResponse", "Subject", - "SubjectActivity", + "SubjectProgress", "SubjectStatus", "SubjectSummary", "Workflow", diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py index 7ae926ef..89921916 100644 --- a/backend/tests/test_durable_sdk.py +++ b/backend/tests/test_durable_sdk.py @@ -5,13 +5,17 @@ import psycopg import pytest +from dbos import DBOS from druks.agents import Agent, AgentOutput +from druks.apps import loader from druks.apps.registry import agents, workflows -from druks.database import configure_session, get_session +from druks.database import configure_session, db_session, get_session, session_scope 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.events.models import Event from druks.models import StoredSubject +from druks.signals import subscribe from druks.testing import init_db from druks.workflows import Gate, Subject, Workflow, step, task from pydantic import BaseModel @@ -192,6 +196,8 @@ async def run(self) -> Decision: class DoubleGateFlow(Workflow): # Two rounds on the same gate — the shape a stale buffered reply would # ghost-resume. + subject = Widget + async def run_multistep(self) -> None: first = await Approve.wait() SINK.append(f"round1:{first.action}") @@ -199,6 +205,9 @@ async def run_multistep(self) -> None: SINK.append(f"round2:{second.action}") replies = [reply.action for reply in self.journal.filter(Approve)] SINK.append(f"gate-journal:{replies}") + SINK.append("gate:completed") + if SINK.count("gate:completed") == 1: + raise asyncio.CancelledError class ConfirmFlow(Workflow): subject = Widget @@ -227,6 +236,18 @@ async def run_multistep(self) -> None: await Approve.wait() SINK.append(f"acct-after:{self.account_id}") + class AnnounceFlow(Workflow): + subject = Widget + + async def run_multistep(self) -> None: + await self.announce("test.revision", revision=1) + await self.announce("test.revision", revision=2) + marker = f"announced:{self.workflow_id}" + SINK.append(marker) + if SINK.count(marker) == 1: + # A worker interruption leaves the run available for recovery. + raise asyncio.CancelledError("Simulated worker interruption") + return ( SampleFlow, AgentFlow, @@ -238,6 +259,7 @@ async def run_multistep(self) -> None: SubjectlessConfirmFlow, ReviewFlow, AttributedFlow, + AnnounceFlow, ScheduledDispatch, RetryingStepFlow, EnqueueInStepFlow, @@ -305,6 +327,7 @@ async def rt(): subjectless_confirm_flow, review_flow, attributed_flow, + announce_flow, scheduled_dispatch, retrying_step_flow, enqueue_in_step_flow, @@ -329,6 +352,7 @@ async def rt(): SubjectlessConfirmFlow=subjectless_confirm_flow, ReviewFlow=review_flow, AttributedFlow=attributed_flow, + AnnounceFlow=announce_flow, ScheduledDispatch=scheduled_dispatch, RetryingStepFlow=retrying_step_flow, EnqueueInStepFlow=enqueue_in_step_flow, @@ -354,6 +378,7 @@ async def rt(): workflows._items.pop("subjectless_confirm_flow", None) workflows._items.pop("review_flow", None) workflows._items.pop("attributed_flow", None) + workflows._items.pop("announce_flow", None) workflows._items.pop("scheduled_dispatch", None) workflows._items.pop("retrying_step_flow", None) workflows._items.pop("enqueue_in_step_flow", None) @@ -485,7 +510,7 @@ async def test_duplicate_replies_to_one_round_collapse(rt): not buffer on the topic and ghost-resume the gate's next round unprompted.""" from sqlalchemy import text - wfid = await rt.DoubleGateFlow.start(subject=None) + wfid = await rt.DoubleGateFlow.start(subject=Widget(id=515151)) parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PARKED) first_asked_at = parked.input_requested_at @@ -518,8 +543,34 @@ async def test_duplicate_replies_to_one_round_collapse(rt): assert "round1:first" in SINK # A fresh reply to the new round is a new key, so it still gets through. + second_asked_at = parked.input_requested_at await parked.resume(action="second") + for _ in range(100): + if "gate:completed" in SINK: + break + await asyncio.sleep(0.1) + assert SINK.count("gate:completed") == 1 + await asyncio.sleep(0.2) + await DBOS.resume_workflow_async(wfid) await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED) + assert SINK.count("gate:completed") == 2 + async with get_session(rt.engine) as session: + events = list( + await session.scalars( + select(Event).where(Event.payload["run"].astext == wfid).order_by(Event.id) + ) + ) + requests = [event for event in events if event.type == "workflow.parked"] + receipts = [ + event for event in events if event.type == "workflow.running" and "result" in event.payload + ] + rounds = [first_asked_at.isoformat(), second_asked_at.isoformat()] + assert [event.payload["input_requested_at"] for event in requests] == rounds + assert [event.payload["input_requested_at"] for event in receipts] == rounds + assert [event.payload["result"] for event in receipts] == [ + {"action": "first"}, + {"action": "second"}, + ] assert "round2:second" in SINK assert "round2:duplicate" not in SINK # Both replies landed on the journal, in reply order. @@ -1222,3 +1273,113 @@ async def test_subjectless_run_emits_no_events(rt): await session.close() assert events == [] + + +async def test_announcements_survive_subscriber_retry_and_workflow_replay(rt): + deliveries = [] + + @subscribe("test.revision", workflow=rt.AnnounceFlow) + async def receive(*, subject: Widget, revision: int) -> None: + async with get_session(rt.engine) as session: + events = list(await session.scalars(select(Event).filter_by(type="test.revision"))) + deliveries.append((revision, len(events))) + if len(deliveries) == 1: + raise RuntimeError("Subscriber unavailable") + + workflow_id = await rt.AnnounceFlow.start(subject=Widget(id=7)) + marker = f"announced:{workflow_id}" + try: + await _wait_for(rt.engine, workflow_id, lambda run: SINK.count(marker) == 1) + assert deliveries == [(1, 1), (1, 1), (2, 2)] + + await DBOS.resume_workflow_async(workflow_id) + await _wait_for(rt.engine, workflow_id, lambda run: run.state == RunState.FINISHED) + assert SINK.count(marker) == 2 + + async with get_session(rt.engine) as session: + events = list( + await session.scalars( + select(Event).filter_by(type="test.revision").order_by(Event.id) + ) + ) + assert [event.payload for event in events] == [ + {"revision": 1, "run": workflow_id, "kind": rt.AnnounceFlow.kind}, + {"revision": 2, "run": workflow_id, "kind": rt.AnnounceFlow.kind}, + ] + assert deliveries == [(1, 1), (1, 1), (2, 2)] + finally: + await DBOS.cancel_workflow_async(workflow_id) + + +async def test_admission_commits_before_the_request_and_deduplicates(rt, monkeypatch): + monkeypatch.setitem(loader._workflow_packages, "test_admission", "admission") + + class AdmissionFlow(Workflow): + __module__ = "test_admission" + subject = Widget + + async def run_multistep(self) -> None: + await DBOS.recv_async("finish") + + subject = Widget(id=7) + workflow_id = "" + try: + with pytest.raises(ValueError, match="Roll back the request"): + async with session_scope(rt.engine): + request_session = db_session() + await request_session.execute(select(Widget).where(Widget.id == 7)) + workflow_id = await AdmissionFlow.start(subject=subject) + assert await AdmissionFlow.start(subject=subject) == workflow_id + assert db_session() is request_session + assert request_session.in_transaction() + + async with get_session(rt.engine) as reader: + events = list( + await reader.scalars( + select(Event).filter_by(type="workflow.scheduled", app="admission") + ) + ) + assert len(events) == 1 + assert events[0].payload == {"run": workflow_id, "kind": AdmissionFlow.kind} + assert events[0].subject_id == "7" + assert events[0].subject_label == "W-7" + raise ValueError("Roll back the request") + + async with get_session(rt.engine) as reader: + events = list( + await reader.scalars( + select(Event).filter_by(type="workflow.scheduled", app="admission") + ) + ) + assert len(events) == 1 + finally: + if workflow_id: + await DBOS.send_async(workflow_id, "done", topic="finish") + await _wait_for(rt.engine, workflow_id, lambda run: run.state == RunState.FINISHED) + workflows._items.pop(AdmissionFlow.kind) + + +async def test_failed_retry_attempts_keep_separate_terminal_records(rt): + class FailingAttempt(Workflow): + subject = Widget + + async def run(self) -> None: + raise FatalError("Source unavailable") + + try: + first_id = await FailingAttempt.start(subject=Widget(id=7)) + await _wait_for(rt.engine, first_id, lambda run: run.state == RunState.FAILED) + async with session_scope(rt.engine): + first_run = await Run.get(first_id) + retry_id = await first_run.retry() + await _wait_for(rt.engine, retry_id, lambda run: run.state == RunState.FAILED) + + assert retry_id != first_id + async with get_session(rt.engine) as reader: + events = list(await reader.scalars(select(Event).filter_by(type="workflow.failed"))) + failures = [event for event in events if event.payload["run"] in {first_id, retry_id}] + assert len(failures) == 2 + assert {event.payload["run"] for event in failures} == {first_id, retry_id} + assert {event.payload["failure"] for event in failures} == {"Source unavailable"} + finally: + workflows._items.pop(FailingAttempt.kind) diff --git a/backend/tests/test_gate_receipt.py b/backend/tests/test_gate_receipt.py index aa1318aa..607853db 100644 --- a/backend/tests/test_gate_receipt.py +++ b/backend/tests/test_gate_receipt.py @@ -1,87 +1,103 @@ +from unittest.mock import AsyncMock + import pytest from dbos import DBOS from dbos._error import DBOSWorkflowCancelledError from druks.durable.exceptions import GateTimeout from druks.durable.models import Run +from druks.events.models import Event from druks.testing import seed_run -from druks.workflows import _park +from druks.workflows import OperatorReply, current_workflow +from druks_field_notes.models import Note from druks_field_notes.workflows import Summarize +from pydantic import ValidationError +from sqlalchemy import select -_ASK = {"presentation": "in_app", "controls": ["approve"], "questions": []} - - -class _ParkedWorkflow: - # The slice of Workflow that _park touches. A subjectless run keeps the emit to - # its facts write (no feed event, no notification) — the receipt path under - # test is exactly that write. - def __init__(self, workflow_id: str) -> None: - self.workflow_id = workflow_id - self._subject = None - - async def _reap_run(self) -> None: - return +_ASK = {"presentation": "in_app", "controls": ["approve", "request_changes"], "questions": []} @pytest.fixture -def _direct_steps(monkeypatch): - # Run each durable step body inline — the test exercises _park's own logic, - # not DBOS checkpointing. - async def _call_through(options, func, *args, **kwargs): - return await func(*args, **kwargs) - - monkeypatch.setattr(DBOS, "run_step_async", _call_through) - - -async def _reload(druks_db, run_id: str) -> Run: - druks_db.expunge_all() - return await druks_db.get(Run, run_id) - - -async def test_answer_stamps_the_receipt_beside_the_gate_clear( - druks_db, _direct_steps, monkeypatch +def direct_steps(monkeypatch): + async def call_through(options, function, *args, **kwargs): + return await function(*args, **kwargs) + + monkeypatch.setattr(DBOS, "run_step_async", call_through) + monkeypatch.setattr("druks.workflows._notify_designated_destination", AsyncMock()) + + +@pytest.fixture(params=["gate", "review"]) +async def request_reply(request, druks_db, direct_steps): + note = await Note.create(body="A request to review") + run = await seed_run(druks_db, kind=Summarize.kind, subject=note) + workflow = Summarize() + workflow._workflow_id = run.id + workflow._subject = note.identity + token = current_workflow.set(workflow) + try: + call = ( + OperatorReply.wait(input_request=_ASK) if request.param == "gate" else workflow.review() + ) + yield run.id, call + finally: + current_workflow.reset(token) + + +async def test_valid_reply_records_the_request_round_before_current_fields_clear( + druks_db, request_reply, monkeypatch ): - run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-answer") + run_id, call = request_reply + monkeypatch.setattr(DBOS, "recv_async", AsyncMock(return_value={"action": "approve"})) - async def _answer(topic, timeout_seconds): - return {"action": "approve"} + reply = await call - monkeypatch.setattr(DBOS, "recv_async", _answer) - payload = await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0) - - assert payload == {"action": "approve"} - run = await _reload(druks_db, run.id) - # The receipt is the round the answer cleared: the same stamp the park - # wrote, which _GATE_CLEARED preserves on the row. - assert run.input_requested_at + assert reply == OperatorReply(action="approve") + druks_db.expunge_all() + run = await druks_db.get(Run, run_id) assert run.answer_parked_at == run.input_requested_at assert not run.input_gate assert not run.input_request + events = list(await druks_db.scalars(select(Event).order_by(Event.id))) + assert [event.type for event in events] == ["workflow.parked", "workflow.running"] + request, receipt = events + for event in events: + assert event.payload["run"] == run_id + assert event.payload["gate"] == "review" + assert event.payload["input_requested_at"] == run.input_requested_at.isoformat() + assert request.payload["input_request"] == _ASK + assert receipt.payload["result"] == {"action": "approve", "answers": {}, "note": ""} + + +@pytest.mark.parametrize("payload", [{"action": "merge"}, {}, None]) +async def test_invalid_reply_or_timeout_records_no_receipt( + druks_db, request_reply, monkeypatch, payload +): + run_id, call = request_reply + monkeypatch.setattr(DBOS, "recv_async", AsyncMock(return_value=payload)) + with pytest.raises(GateTimeout if payload is None else ValidationError): + await call -async def test_timeout_never_writes_the_receipt(druks_db, _direct_steps, monkeypatch): - run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-timeout") - - async def _lapse(topic, timeout_seconds): - return None - - monkeypatch.setattr(DBOS, "recv_async", _lapse) - with pytest.raises(GateTimeout): - await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0) - - run = await _reload(druks_db, run.id) + druks_db.expunge_all() + run = await druks_db.get(Run, run_id) assert not run.answer_parked_at + assert run.input_gate == "review" + assert run.input_request == _ASK assert run.input_requested_at + events = list(await druks_db.scalars(select(Event))) + assert [event.type for event in events] == ["workflow.parked"] -async def test_cancel_never_writes_the_receipt(druks_db, _direct_steps, monkeypatch): - run = await seed_run(druks_db, kind=Summarize.kind, run_id="run-receipt-cancel") +async def test_cancel_records_no_receipt(druks_db, request_reply, monkeypatch): + run_id, call = request_reply + monkeypatch.setattr( + DBOS, "recv_async", AsyncMock(side_effect=DBOSWorkflowCancelledError(run_id)) + ) - async def _cancelled(topic, timeout_seconds): - raise DBOSWorkflowCancelledError(run.id) - - monkeypatch.setattr(DBOS, "recv_async", _cancelled) with pytest.raises(DBOSWorkflowCancelledError): - await _park(_ParkedWorkflow(run.id), "review", _ASK, ttl_seconds=1.0) + await call - run = await _reload(druks_db, run.id) + druks_db.expunge_all() + run = await druks_db.get(Run, run_id) assert not run.answer_parked_at + events = list(await druks_db.scalars(select(Event))) + assert [event.type for event in events] == ["workflow.parked"] diff --git a/backend/tests/test_operator_stops.py b/backend/tests/test_operator_stops.py new file mode 100644 index 00000000..5e7939fa --- /dev/null +++ b/backend/tests/test_operator_stops.py @@ -0,0 +1,70 @@ +import pytest +from druks.contrib.software_factory.workflows import Build +from druks.db import db_session +from druks.durable.enums import WorkflowEvent +from druks.events.models import Event +from druks.testing import seed_run +from druks_field_notes.models import Note +from druks_field_notes.workflows import Summarize +from software_factory.factories import make_test_work_item +from sqlalchemy import select + + +async def test_operator_stop_records_its_reason_and_exact_run_once(druks_client, druks_db): + note = await Note.create(body="Stop this work") + run = await seed_run(druks_db, kind=Summarize.kind, subject=note) + + response = await druks_client.post( + f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"} + ) + assert response.status_code == 200 + assert response.json() == {"run": run.id, "result": "cancelled"} + + repeated = await druks_client.post( + f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"} + ) + assert repeated.json() == {"run": run.id, "result": "already_cancelled"} + events = list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED))) + assert len(events) == 1 + assert events[0].payload == {"run": run.id, "kind": Summarize.kind, "reason": "Wrong source"} + assert events[0].app == "field_notes" + assert events[0].subject_id == str(note.id) + assert events[0].subject_label == note.label + + +async def test_failed_operator_cancellation_records_no_stop(druks_client, druks_db, monkeypatch): + note = await Note.create(body="Cancellation failed") + run = await seed_run(druks_db, kind=Summarize.kind, subject=note) + + async def unavailable(workflow_id: str) -> None: + raise RuntimeError("Cancellation unavailable") + + monkeypatch.setattr("dbos.DBOS.cancel_workflow_async", unavailable) + with pytest.raises(RuntimeError, match="Cancellation unavailable"): + await druks_client.post(f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"}) + + assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED))) + + +@pytest.mark.parametrize("state", ["failed", "finished"]) +async def test_inactive_run_has_no_operator_stop(druks_client, druks_db, state): + note = await Note.create(body="Finished work") + run = await seed_run(druks_db, kind=Summarize.kind, subject=note, state=state) + + response = await druks_client.post( + f"/api/runs/{run.id}/cancel", json={"reason": "Wrong source"} + ) + + assert response.status_code == 409 + assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED))) + + +@pytest.mark.parametrize("reason", ["Pull request merged", "Pull request closed"]) +async def test_factory_cleanup_creates_no_operator_stop(druks_db, reason): + db_session.registry.set(druks_db) + item = await make_test_work_item(repo="owner/repo", title="Completed work") + await seed_run(druks_db, kind=Build.kind, subject=item) + + await Build.cancel(item, failure=reason) + + assert not list(await druks_db.scalars(select(Event).filter_by(type=WorkflowEvent.CANCELLED))) diff --git a/backend/tests/test_run_state.py b/backend/tests/test_run_state.py index 75a79a43..c1e96d2f 100644 --- a/backend/tests/test_run_state.py +++ b/backend/tests/test_run_state.py @@ -162,7 +162,11 @@ async def _raises(**_: object) -> None: run.id, RunState.PARKED, subject={"type": "work_item", "id": item.id}, - facts={"input_gate": "review_work", "input_request": {"label": "Review"}}, + facts={ + "input_gate": "review_work", + "input_request": {"label": "Review"}, + "input_requested_at": datetime.now(UTC), + }, ) ambient_session().expunge_all() @@ -364,11 +368,10 @@ async def body() -> None: @pytest.mark.asyncio async def test_announce_carries_the_runs_routing(druks_db): - # The body states its facts; the platform injects what subscribers filter on, - # and the publish rides its own named checkpoint — the boundary that keeps a - # recovery replay from re-firing it. - workflow = Workflow() - workflow._subject = {"type": "note", "id": 7} + note, run = await _item_and_run(druks_db, "running") + workflow = Summarize() + workflow._workflow_id = run.id + workflow._subject = note.identity received = [] checkpoints = [] @@ -383,8 +386,12 @@ async def run_inline(options, fn): with mock.patch("druks.workflows.DBOS.run_step_async", side_effect=run_inline): await workflow.announce("test.announced", pr_number=12) - assert received == [({"type": "note", "id": 7}, {"pr_number": 12})] - assert checkpoints == ["test.announced"] + assert received == [(note.identity, {"pr_number": 12})] + assert checkpoints == ["test.announced", "test.announced:propagate"] + event = (await ambient_session().scalars(select(Event).filter_by(type="test.announced"))).one() + assert event.app == "field_notes" + assert event.subject_label == note.label + assert event.payload == {"pr_number": 12, "run": run.id, "kind": workflow.kind} @pytest.mark.asyncio diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index c32a7300..5782e563 100644 --- a/docs/writing-an-app.md +++ b/docs/writing-an-app.md @@ -241,15 +241,43 @@ Two rules: ### Announcing domain events -If another component must react to a body action, announce the action: +Announce a domain fact from the workflow body: ```python await self.announce("pr.opened", pr_number=delivery.pr_number, branch=delivery.branch) ``` -The platform routes it to subscribers that filter on your workflow and subject. -The publication is a durable checkpoint. Recovery does not publish it again. -Announce from the body, not inside a `@step`. +Druks records the event in one checkpoint. It notifies subscribers in a second +checkpoint. A subscriber retry cannot insert the completed event again. Recovery +reuses completed checkpoints. An interrupted operation can run again, so +subscribers must remain idempotent. Call this method outside a `@step`. + +A domain method announces through its subject: + +```python +from druks.db import StoredSubject +from sqlalchemy.orm import Mapped + + +class Report(StoredSubject): + __tablename__ = "night_watch_reports" + + published_url: Mapped[str | None] + + async def publish(self, url: str) -> None: + if self.published_url != url: + self.published_url = url + await self.announce("report.published", url=url) +``` + +`Subject` and `StoredSubject` both supply `announce()`. Druks gets the owner from +the registered app package. The call records the subject identity, its current +label, and the supplied facts. It then notifies subscribers in the same +transaction as the domain change. A rollback removes the change and its event. +The app must prevent duplicate domain changes on webhook redelivery. + +Authors supply no app ID, run ID, timestamp, or session. Frontend code owns the +Activity wording. ### Schedules and settings @@ -733,8 +761,26 @@ Druks serves the same `/api/night_watch/repository` surface for both subject types. This surface contains a board, detail pages, and a live stream. Druks mounts it for each declared subject. Each response contains your summary, run status, timeline, agent calls, artifacts, and active question. Override -`get_subject_activity()` only to add transient app detail, such as -"Building sandbox VM…". +`get_subject_progress()` to add labeled live detail: + +```python +from druks.apps import App +from druks.db import StoredSubject +from druks.workflows import SubjectProgress + + +class NightWatch(App): + name = "night_watch" + + @classmethod + async def get_subject_progress(cls, subject: StoredSubject) -> SubjectProgress | None: + if await subject.get_phase() == "sandbox_building": + return SubjectProgress(label="Building sandbox…", kind="infra") +``` + +The response carries this detail in `progress`. `Subject.get_phase()` and +`durable.reads.get_subject_phase()` return the raw step string. Activity names +recorded history. Pass the subject instance to each component that requires one. This includes a workflow start, gate answer, or event: diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index 0907b643..2f73a333 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -80,7 +80,7 @@ export interface SubjectStatus { // The live sub-phase a running run pushes ("Provisioning sandbox VM…", "Working…") — // finer than the lifecycle status; null unless something is actively running. -export interface SubjectActivity { +export interface SubjectProgress { label: string kind: string } @@ -160,12 +160,12 @@ export interface SubjectRow { // A subject's full read view: domain summary, status, the platform timeline // (the subject's runs, oldest first, each with its agent calls), and the -// app's optional live activity (the running sub-phase). +// app's optional live progress (the running sub-phase). export interface SubjectResponse { summary: S status: SubjectStatus timeline: RunSummary[] - activity?: SubjectActivity | null + progress?: SubjectProgress | null } export interface ArtifactFile { diff --git a/frontend/src/apps/software_factory/WorkItemPage.tsx b/frontend/src/apps/software_factory/WorkItemPage.tsx index 9c18cdaf..94119ede 100644 --- a/frontend/src/apps/software_factory/WorkItemPage.tsx +++ b/frontend/src/apps/software_factory/WorkItemPage.tsx @@ -10,7 +10,7 @@ import type { AgentCallSummary, RunState, RunSummary, - SubjectActivity, + SubjectProgress, SubjectStatus, } from '../../api/types' import { DetailLayout } from '../../components/DetailLayout' @@ -190,7 +190,7 @@ function WorkItemView({ data }: { data: WorkItemDetail }) { /> setSelected(id)} /> @@ -308,12 +308,12 @@ function InfoPanel({ function TimelinePanel({ runs, - activity, + progress, selection, onSelect, }: { runs: RunSummary[] - activity?: SubjectActivity | null + progress?: SubjectProgress | null selection: Selection | null onSelect: (id: string) => void }) { @@ -331,7 +331,7 @@ function TimelinePanel({ @@ -343,12 +343,12 @@ function TimelinePanel({ function RunRow({ run, - activity, + progress, selection, onSelect, }: { run: RunSummary - activity?: SubjectActivity | null + progress?: SubjectProgress | null selection: Selection | null onSelect: (id: string) => void }) { @@ -357,7 +357,7 @@ function RunRow({ // A single call duplicates the run's own row (same label, same ledger) — // fold it into the parent instead of showing both. const collapseCalls = run.agentCalls.length <= 1 - const subtitle = runSubLine(run, activity, collapseCalls) + const subtitle = runSubLine(run, progress, collapseCalls) return (
◍
-
{data.activity?.label ?? 'Starting up…'}
+
{data.progress?.label ?? 'Starting up…'}
no agent call yet — the transcript begins once the agent starts
diff --git a/frontend/src/apps/software_factory/statusLine.ts b/frontend/src/apps/software_factory/statusLine.ts index 85e8a9b6..9776f196 100644 --- a/frontend/src/apps/software_factory/statusLine.ts +++ b/frontend/src/apps/software_factory/statusLine.ts @@ -1,5 +1,5 @@ import type { PRResolution } from './api' -import type { RunSummary, SubjectActivity, SubjectStatus } from '../../api/types' +import type { RunSummary, SubjectProgress, SubjectStatus } from '../../api/types' // Build's status-line copy, composed from the platform's status facts — the backend // ships data; the app owns its own vocabulary. @@ -51,10 +51,10 @@ const STATE_LABEL: Record = { } // What a run row says it is doing. A folded-in step names itself, since its own -// row isn't there to; the activity covers what the timeline can't show at all. +// row isn't there to; the progress covers what the timeline can't show at all. export function runSubLine( run: RunSummary, - activity: SubjectActivity | null | undefined, + progress: SubjectProgress | null | undefined, collapsed: boolean, ): string { if (run.state === 'failed' && run.failure) { @@ -66,7 +66,7 @@ export function runSubLine( if (run.state === 'running') { const step = collapsed && run.agentCalls.find((call) => call.status === 'running') if (step) return step.label - if (activity) return activity.label + if (progress) return progress.label } return STATE_LABEL[run.state] ?? run.state } diff --git a/frontend/src/pages/SubjectPage.tsx b/frontend/src/pages/SubjectPage.tsx index ef7b482d..62d3f8e4 100644 --- a/frontend/src/pages/SubjectPage.tsx +++ b/frontend/src/pages/SubjectPage.tsx @@ -115,7 +115,7 @@ function SubjectDetail({ {value} ))} - {data.activity && {data.activity.label}} + {data.progress && {data.progress.label}} {selectedRun && !runs.some((run) => run.id === selectedRun) && (

This run does not belong to this subject or is no longer available.