From 65ba44fe21ec52d629cedcf8ec291366fce65929 Mon Sep 17 00:00:00 2001 From: Krzysztof Socha Date: Tue, 10 Nov 2026 11:48:10 +0100 Subject: [PATCH] DRU-1146 -- A workflow reads the cost of its own run --- backend/druks/durable/models.py | 5 +++++ backend/druks/workflows.py | 10 ++++++++++ backend/tests/test_cost_capture.py | 14 ++++++++++++++ docs/writing-an-app.md | 13 +++++++++++++ 4 files changed, 42 insertions(+) diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index ac4f127f..2f45050f 100644 --- a/backend/druks/durable/models.py +++ b/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. diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index ef593dd4..665b148d 100644 --- a/backend/druks/workflows.py +++ b/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: diff --git a/backend/tests/test_cost_capture.py b/backend/tests/test_cost_capture.py index e2e90012..eaf7919f 100644 --- a/backend/tests/test_cost_capture.py +++ b/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}) diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index 3bb982b2..b87a4d3b 100644 --- a/docs/writing-an-app.md +++ b/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: