DRU-1146 -- A workflow reads the cost of its own run

backend/druks/durable/models.py

@@ -694,6 +694,11 @@ async def lookup_running(
             )
         )
 
+    @classmethod
+    async def get_cost_for_run(cls, session: AsyncSession, run_id: str) -> float:
+        stmt = select(func.sum(cls.cost_usd)).where(cls.run_id == run_id)
+        return await session.scalar(stmt) or 0.0
+
     @classmethod
     async def list_for_run(cls, session: AsyncSession, run_id: str) -> list["AgentCall"]:
         # Execution order — the same order Run.agent_calls loads.

backend/druks/workflows.py

@@ -928,6 +928,16 @@ async def notify() -> None:
 
         await DBOS.run_step_async(StepOptions(name=f"{topic}:propagate", **_IO_RETRIES), notify)
 
+    async def get_cost(self) -> float:
+        """What this run's agent calls have cost so far, in US dollars, as recorded. A
+        checkpoint: a replay reads the cost the run read the first time."""
+
+        async def read() -> float:
+            async with step_session() as session:
+                return await AgentCall.get_cost_for_run(session, self.workflow_id)
+
+        return await DBOS.run_step_async(StepOptions(name="get_cost", **_IO_RETRIES), read)
+
     async def review(
         self, *, questions: list[BaseModel] | None = None, context: str = ""
     ) -> OperatorReply:

backend/tests/test_cost_capture.py

@@ -272,6 +272,20 @@ async def test_record_agent_run_cost_noop_when_empty(druks_db):
     assert fetched.cost_metadata is None
 
 
+async def test_a_runs_cost_sums_its_calls_and_counts_a_missing_cost_as_zero(druks_db):
+    note = await Note.create(body="run cost")
+    run = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+    other = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+    empty = await seed_run(druks_db, kind=Summarize.kind, subject=note)
+    for on, cost in ((run, 0.25), (run, 1.5), (run, None), (other, 9.0)):
+        call = await seed_call(druks_db, on, "summarize")
+        call.cost_usd = cost
+    await druks_db.flush()
+
+    assert await AgentCall.get_cost_for_run(druks_db, run.id) == 1.75
+    assert await AgentCall.get_cost_for_run(druks_db, empty.id) == 0.0
+
+
 def test_cost_file_format_is_loadable_json(tmp_path: Path):
     write_cost(tmp_path, cost_usd=0.42, metadata={"input_tokens": 50})
 

docs/writing-an-app.md

@@ -360,6 +360,19 @@ The app must prevent duplicate domain changes on webhook redelivery.
 Authors supply no app ID, run ID, timestamp, or session. Frontend code owns the
 wording.
 
+### Read what the run has cost
+
+Read the run's cost so far from the workflow body, for example for a summary:
+
+```python
+cost = await self.get_cost()
+```
+
+It is the sum, in US dollars, of the costs that the run's agent calls recorded. A
+call that recorded no cost counts as zero. The read is a checkpoint, so a replay
+returns the cost the run read the first time. Inside a `@step` or `run()`, the
+read shares that checkpoint.
+
 ### Schedules and settings
 
 Set `every` to declare a cron: