czpython · czpython · Sep 9, 2026 · Sep 9, 2026
@@ -154,6 +154,9 @@ class PlanOutput(AgentOutput):
     def get_artifact(self) -> dict[str, str]:
         return {"kind": "markdown", "title": "Implementation plan", "content": self.plan_markdown}
 
+    def get_activity(self) -> dict[str, str]:
+        return {"kind": "plan.prepared"}
+
     def to_result(self) -> PlanData:
         return PlanData(
             plan_markdown=self.plan_markdown,
@@ -173,6 +176,9 @@ class ContractRevisionOutput(AgentOutput):
     def get_artifact(self) -> dict[str, str]:
         return {"kind": "markdown", "title": "Implementation plan", "content": self.plan_markdown}
 
+    def get_activity(self) -> dict[str, str]:
+        return {"kind": "plan.revised"}
+
     def to_result(self) -> PlanData:
         # A revision resolves the questions, so none carry over;
         # implementation_instructions ride the prompt, not the plan artifact.
@@ -274,6 +280,25 @@ class ReviewReport(AgentOutput):
     findings: list[FindingOutput]
     context_repos: list[str]
 
+    def get_artifact(self) -> dict[str, str]:
+        sections = [f"Decision: {self.decision}", self.summary]
+        for finding in self.findings:
+            sections.extend([f"## {finding.summary}", finding.evidence])
+            if finding.path:
+                location = finding.path
+                if finding.line:
+                    lines = (
+                        f"{finding.start_line}-{finding.line}"
+                        if finding.start_line and finding.start_line != finding.line
+                        else str(finding.line)
+                    )
+                    location = f"{location}:{lines}"
+                sections.append(f"Source: `{location}`")
+        return {"kind": "markdown", "title": "Review", "content": "\n\n".join(sections)}
+
+    def get_activity(self) -> dict[str, str]:
+        return {"kind": "review.completed", "summary": self.summary}
+
 
 class EvalCheckOutput(AgentOutput):
     name: str

@@ -288,7 +288,10 @@ async def get_for_pr(
             found = (await db_session().scalars(stmt)).first()
             if found:
                 return found
-        return await cls.get_for_branch(repo=repo, branch=branch) if branch else None
+        if branch:
+            found = await cls.get_for_branch(repo=repo, branch=branch)
+            if found and (not pr_number or not found.pr_number or found.pr_number == pr_number):
+                return found
 
     @classmethod
     async def get_for_branch(cls, *, repo: str, branch: str) -> "WorkItem | None":
@@ -309,13 +312,17 @@ async def start_attempt(self) -> None:
         await db_session().flush()
 
     async def resolve(self, *, merged: bool, at: datetime) -> None:
-        # cycle: the app imports this module at file scope.
-        import druks.contrib.software_factory.app as software_factory_app
-
         self.resolution = "merged" if merged else "closed"
         self.resolved_at = at
         self.updated_at = Base.utc_now()
-        await software_factory_app.SoftwareFactory.record_event(type=self.resolution, subject=self)
+        await self.announce(self.resolution)
+        await db_session().flush()
+
+    async def stop(self) -> None:
+        """End the attempt after an operator stop, without a GitHub outcome."""
+        self.resolution = "closed"
+        self.resolved_at = Base.utc_now()
+        self.updated_at = self.resolved_at
         await db_session().flush()
 
     async def ship(self) -> None:

@@ -4,7 +4,6 @@
 from druks.contrib.software_factory.models import ProjectRepo, WorkItem
 from druks.contrib.software_factory.ticketing.enums import TicketStatus
 from druks.contrib.software_factory.workflows import Build, Profile, PullRequestReview
-from druks.db import Base
 from druks.signals import subscribe
 from druks.workflows import WorkflowEvent
 
@@ -18,7 +17,7 @@ async def new_build_claims_the_item(*, subject: WorkItem, **_: object) -> None:
 async def cancelled_build_settles_the_item(*, subject: WorkItem, **_: object) -> None:
     """An operator cancellation explicitly abandons the work item."""
     if not subject.resolution:
-        await subject.resolve(merged=False, at=Base.utc_now())
+        await subject.stop()
 
 
 @subscribe("pr.opened", workflow=Build)

@@ -157,10 +157,9 @@ async def dispatch(cls, *, ticket: dict) -> str | None:
         try:
             await Github.get()
         except ServiceNotConnectedError as error:
-            # A raise would 5xx the tracker's webhook and put the delivery into
-            # provider redelivery; the delivery itself succeeded. Log the
-            # Connect GitHub direction and stand down without starting.
+            # The tracker delivery succeeded. A raise would request another delivery.
             logger.info("Ticket %s cannot start a build: %s", ticket["identifier"], error)
+            await item.announce("build.rejected", reason=str(error))
             return
         email = ticket["assignee_email"]
         assignee = await Account.get_for_username(email.strip()) if email else None

@@ -0,0 +1,162 @@
+from datetime import UTC, datetime
+from unittest.mock import AsyncMock
+
+import pytest
+from conftest import installation_key
+from druks.contrib.software_factory.contracts import (
+    ContractRevisionOutput,
+    FindingOutput,
+    PlanOutput,
+    ReviewReport,
+)
+from druks.contrib.software_factory.datastructures import PullRequest
+from druks.contrib.software_factory.models import WorkItem
+from druks.contrib.software_factory.subscribers import pr_close_settles_the_item
+from druks.contrib.software_factory.workflows import Build, PullRequestReview
+from druks.database import db_session
+from druks.durable.models import AgentCall, Artifact
+from druks.events.models import Event
+from druks.testing import seed_run
+from sqlalchemy import select
+
+from software_factory.factories import make_test_work_item
+
+
+def test_plan_outputs_declare_their_saved_results():
+    plan = PlanOutput(
+        plan_markdown="# Plan",
+        acceptance_criteria=[],
+        questions=[],
+        rejected_approaches=[],
+        confidence="high",
+        assignee_github_login=None,
+    )
+    revision = ContractRevisionOutput(
+        plan_markdown="# Revised plan",
+        acceptance_criteria=[],
+        implementation_instructions="Build it.",
+    )
+    assert plan.get_activity() == {"kind": "plan.prepared"}
+    assert revision.get_activity() == {"kind": "plan.revised"}
+    assert plan.get_artifact()["content"] == "# Plan"
+    assert revision.get_artifact()["content"] == "# Revised plan"
+
+
+async def test_review_result_belongs_to_the_identity_only_pull_request(druks_db, tmp_path):
+    db_session.registry.set(druks_db)
+    subject = PullRequest.get("acme/widget", 42)
+    run = await seed_run(druks_db, kind=PullRequestReview.kind, subject=subject)
+    key = await installation_key()
+    call = AgentCall(
+        id="factory-review",
+        run_id=run.id,
+        agent="software_factory.review_pull_request",
+        model="test",
+        sandbox_host_id="test",
+        api_key_id=key.id,
+    )
+    druks_db.add(call)
+    await druks_db.flush()
+    druks_db.expunge_all()
+    report = ReviewReport(
+        decision="request_changes",
+        summary="The write can lose data.",
+        context_repos=[],
+        findings=[
+            FindingOutput(
+                severity="high",
+                summary="Keep the transaction open",
+                evidence="The commit precedes the write.",
+                path="backend/write.py",
+                line=12,
+                start_line=10,
+            )
+        ],
+    )
+    for _ in range(2):
+        await Artifact.record(
+            call_id=call.id,
+            call_dir=tmp_path,
+            activity=report.get_activity(),
+            **report.get_artifact(),
+        )
+    artifact = await Artifact.get_for_call(call.id)
+    content = (tmp_path / artifact.path).read_text()
+    assert "request_changes" in content
+    assert "The write can lose data." in content
+    assert "## Keep the transaction open" in content
+    assert "The commit precedes the write." in content
+    assert "backend/write.py:10-12" in content
+    events = list(await druks_db.scalars(select(Event)))
+    assert len(events) == 1
+    event = events[0]
+    assert event.type == "review.completed"
+    assert event.app == "software_factory"
+    assert (event.subject_type, event.subject_id) == ("pull_request", "acme/widget#42")
+    assert event.payload["artifact_id"] == artifact.id
+    assert event.payload["agent_call_id"] == call.id
+    assert event.payload["run"] == run.id
+    assert subject.url == "https://github.com/acme/widget/pull/42"
+
+
+async def test_owner_outcome_and_announcement_roll_back_together(druks_db, monkeypatch):
+    db_session.registry.set(druks_db)
+    item = await make_test_work_item(repo="acme/widget", title="Atomic outcome")
+    item_id = item.id
+    monkeypatch.setattr(Event, "emit", AsyncMock(side_effect=RuntimeError("Event insert failed")))
+    with pytest.raises(RuntimeError, match="Event insert failed"):
+        async with druks_db.begin_nested():
+            await item.resolve(merged=True, at=datetime.now(UTC))
+    druks_db.expunge_all()
+    assert not (await WorkItem.get(item_id)).resolution
+    assert not list(await druks_db.scalars(select(Event)))
+
+
+async def test_stale_pr_on_a_reused_branch_cannot_resolve_the_current_attempt(druks_db):
+    item = await make_test_work_item(repo="acme/widget", title="Current attempt")
+    await item.update(pr_number=43, branch="agent/current")
+    await pr_close_settles_the_item(
+        repo=item.repo,
+        pr_number=42,
+        payload={"branch": item.branch, "merged": True, "resolved_at": datetime.now(UTC)},
+    )
+    assert not item.resolution
+
+
+@pytest.mark.parametrize("pr_number", [None, 42])
+async def test_operator_stop_records_no_owner_close(druks_db, druks_client, pr_number):
+    db_session.registry.set(druks_db)
+    item = await make_test_work_item(repo="acme/widget", title="Stopped work")
+    await item.update(pr_number=pr_number, branch="agent/stopped")
+    run = await seed_run(druks_db, kind=Build.kind, subject=item)
+    for _ in range(2):
+        response = await druks_client.post(
+            f"/api/runs/{run.id}/cancel", json={"reason": "Operator stopped work"}
+        )
+        assert response.status_code == 200
+    events = list(await druks_db.scalars(select(Event).where(Event.subject_id == str(item.id))))
+    assert [event.type for event in events] == ["workflow.cancelled"]
+    druks_db.expunge_all()
+    assert (await WorkItem.get(item.id)).resolution == "closed"
+
+
+@pytest.mark.parametrize("state", ["parked", "failed"])
+async def test_owner_merge_records_once_without_an_operator_stop(druks_db, state):
+    item = await make_test_work_item(repo="acme/widget", title="Merged work")
+    await item.update(pr_number=42, branch="agent/merged")
+    await seed_run(
+        druks_db,
+        kind=Build.kind,
+        subject=item,
+        state=state,
+        input_gate="review_work" if state == "parked" else None,
+    )
+    for _ in range(2):
+        await pr_close_settles_the_item(
+            repo=item.repo,
+            pr_number=42,
+            payload={"branch": item.branch, "merged": True, "resolved_at": datetime.now(UTC)},
+        )
+    events = list(await druks_db.scalars(select(Event).where(Event.subject_id == str(item.id))))
+    assert [event.type for event in events] == ["merged"]
+    assert item.resolution == "merged"
@@ -2,8 +2,10 @@
 
 from conftest import connect_service
 from druks.contrib.software_factory.workflows import Build
+from druks.events.models import Event
 from druks.signals import publish
 from druks.testing import seed_run
+from sqlalchemy import select
 
 from software_factory.factories import make_test_work_item
 
@@ -74,6 +76,13 @@ async def fake_start(cls, **kwargs):
 
     assert result is None
     assert not started
+    events = list(await druks_db.scalars(select(Event)))
+    assert len(events) == 1
+    assert events[0].type == "build.rejected"
+    assert events[0].app == "software_factory"
+    assert events[0].subject_id == str(item.id)
+    assert "not connected" in events[0].payload["reason"]
+    assert "run" not in events[0].payload
     assert any("not connected" in record.getMessage() for record in caplog.records)
 
 

@@ -1,11 +1,17 @@
 import type { ReactNode } from 'react'
+import type { FeedItem, InputRequest } from '../api/types'
 
 export interface AppRoute {
   /** A wouter pattern under the router base, such as /notes/:id. */
   path: string
   render: (params: Record<string, string>) => ReactNode
 }
 
+export type ActivityEvent = Pick<FeedItem, 'kind' | 'workflow'> & {
+  gate?: string | null
+  inputRequest?: InputRequest | null
+}
+
 export interface AppUI {
   name: string
   /** The app's default destination. Defaults to /<name>. */
@@ -18,6 +24,7 @@ export interface AppUI {
   // Where a feed row about one of this app's subjects navigates. The shell knows
   // an app has subjects, never where its pages put them.
   subjectPath?: (subject: { type: string; id: string }, target?: SubjectTarget) => string | undefined
+  activityLabel?: (event: ActivityEvent) => string | undefined
   parentPath?: (location: string) => string | undefined
 }
 

@@ -0,0 +1,39 @@
+import { describe, expect, it } from 'vitest'
+import { activityLabel } from './activity'
+import { getAppUI } from '../registry'
+import { eventLine } from '../../lib/feed'
+import './ui'
+
+describe('Factory Activity', () => {
+  it.each([
+    ['workflow.scheduled', 'Build queued'],
+    ['plan.prepared', 'Plan prepared'],
+    ['plan.revised', 'Plan revised'],
+    ['pr.opened', 'Pull request opened'],
+    ['review.completed', 'Review completed'],
+    ['merged', 'Pull request merged'],
+    ['closed', 'Pull request closed'],
+    ['workflow.failed', 'Build failed'],
+    ['build.rejected', 'Build could not start'],
+    ['workflow.cancelled', 'Build stopped'],
+  ])('formats %s through the app registry', (kind, label) => {
+    expect(eventLine({ id: 'event:1', seq: 1, at: '2026-09-09T12:00:00Z',
+      kind, app: 'software_factory', workflow: 'software_factory.build' }).label).toBe(label)
+  })
+
+  it('uses the gate and request to name decisions', () => {
+    const event = { kind: 'workflow.parked', workflow: 'software_factory.build' }
+    expect(activityLabel({ ...event, gate: 'review' })).toBe('Plan review requested')
+    expect(activityLabel({ ...event, gate: 'review_work' })).toBe('Implementation review requested')
+    expect(activityLabel({ ...event, gate: 'review', inputRequest: {
+      presentation: 'in_app', questions: [{ id: 'q', prompt: 'Which source?', options: [] }],
+    } })).toBe('Clarification requested')
+    expect(activityLabel({ ...event, kind: 'workflow.running', gate: 'review' })).toBe('Response received')
+    expect(activityLabel({ ...event, kind: 'workflow.running' })).toBeUndefined()
+  })
+
+  it('links an identity-only pull request to its owner', () => {
+    expect(getAppUI('software_factory')?.subjectPath?.({ type: 'pull_request', id: 'acme/widget#42' }))
+      .toBe('https://github.com/acme/widget/pull/42')
+  })
+})
@@ -0,0 +1,31 @@
+import type { ActivityEvent } from '../registry'
+
+const TOPICS: Record<string, string> = {
+  'plan.prepared': 'Plan prepared',
+  'plan.revised': 'Plan revised',
+  'pr.opened': 'Pull request opened',
+  'review.completed': 'Review completed',
+  merged: 'Pull request merged',
+  closed: 'Pull request closed',
+  'build.rejected': 'Build could not start',
+}
+
+export function activityLabel(event: ActivityEvent): string | undefined {
+  if (TOPICS[event.kind]) return TOPICS[event.kind]
+  if (event.workflow === 'software_factory.build') {
+    switch (event.kind) {
+      case 'workflow.scheduled': return 'Build queued'
+      case 'workflow.failed': return 'Build failed'
+      case 'workflow.cancelled': return 'Build stopped'
+      case 'workflow.parked':
+        if (event.gate === 'review_work') return 'Implementation review requested'
+        if (event.gate === 'review') {
+          return event.inputRequest?.questions?.length ? 'Clarification requested' : 'Plan review requested'
+        }
+        return 'Review requested'
+      case 'workflow.running':
+        if (event.gate) return 'Response received'
+    }
+  }
+  return undefined
+}