From d32e980d96cc3974d600ce9c2ad93f17e993b7c9 Mon Sep 17 00:00:00 2001 From: Pravin Kamble Date: Tue, 3 Nov 2026 13:57:08 +0530 Subject: [PATCH] Handle Claude spend limits and quota waits --- backend/druks/agents.py | 6 +- backend/druks/durable/activity.py | 10 +- backend/druks/durable/reads.py | 7 +- backend/druks/durable/schemas.py | 5 +- backend/druks/harnesses/base.py | 3 +- backend/druks/harnesses/claude.py | 7 +- backend/druks/harnesses/codex.py | 12 +- backend/druks/harnesses/opencode.py | 34 ++-- backend/druks/harnesses/pi.py | 31 ++-- backend/tests/conftest.py | 1 + .../software_factory/test_api_work_items.py | 27 ++- backend/tests/test_agents.py | 6 + backend/tests/test_cost_capture.py | 168 +++++++++++++++++- backend/tests/test_opencode.py | 24 ++- backend/tests/test_pi.py | 55 +++++- backend/tests/test_sandboxed_harness.py | 9 + frontend/src/api/types.ts | 7 +- .../apps/software_factory/WorkItemPage.tsx | 36 +++- .../apps/software_factory/statusLine.test.ts | 6 + .../src/apps/software_factory/statusLine.ts | 4 +- frontend/src/lib/phase.ts | 19 +- frontend/src/pages/SubjectPage.tsx | 4 +- 22 files changed, 413 insertions(+), 68 deletions(-) diff --git a/backend/druks/agents.py b/backend/druks/agents.py index 069f5ccca..a14a3cc1f 100644 --- a/backend/druks/agents.py +++ b/backend/druks/agents.py @@ -5,7 +5,7 @@ from collections.abc import Awaitable, Callable from contextvars import ContextVar from dataclasses import dataclass, field -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any from dbos import DBOS, StepOptions @@ -284,6 +284,10 @@ async def _quota_retry_wait() -> float: if wait_seconds > _QUOTA_MAX_WAIT_SECONDS: raise wait_seconds += random.uniform(60, 120) + await set_run_phase( + "waiting_for_quota", + ends_at=datetime.now(UTC) + timedelta(seconds=wait_seconds), + ) else: # Permanent or unclassified — never retried. raise diff --git a/backend/druks/durable/activity.py b/backend/druks/durable/activity.py index 0487d52c8..295c12c33 100644 --- a/backend/druks/durable/activity.py +++ b/backend/druks/durable/activity.py @@ -1,13 +1,21 @@ +from datetime import datetime + from dbos import DBOS -async def set_run_phase(phase: str) -> None: +async def set_run_phase(phase: str, *, ends_at: datetime | None = None) -> None: # The live infra/agent phase, pushed as a DBOS event (transient by nature — # overwritten each phase, gone when the run ends). NOT a durable column. # A chat or browser box provisions outside a run: nothing shows its phase. if DBOS.workflow_id: await DBOS.set_event_async("phase", phase) + if ends_at: + await DBOS.set_event_async("phase_ends_at", ends_at) async def get_run_phase(run_id: str) -> str | None: return await DBOS.get_event_async(run_id, "phase", timeout_seconds=0) + + +async def get_run_phase_ends_at(run_id: str) -> datetime | None: + return await DBOS.get_event_async(run_id, "phase_ends_at", timeout_seconds=0) diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py index a075c5c54..b726fb675 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -10,7 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from druks.database import session_scope -from druks.durable.activity import get_run_phase +from druks.durable.activity import get_run_phase, get_run_phase_ends_at from druks.durable.exceptions import AgentCallNotFound from druks.durable.live import keepalive_comment, serialize_model_event @@ -126,11 +126,14 @@ async def get_subject_response( # eager-loaded here. runs = await Run.list_for_subject(session, subject_type, subject_id, include_calls=True) latest = runs[0] if runs else None + phase = await get_run_phase(latest.id) if latest and latest.is_running else None + ends_at = await get_run_phase_ends_at(latest.id) if phase else None return SubjectResponse( summary=summary, status=await _status(latest), timeline=await _timeline(runs), - phase=await get_run_phase(latest.id) if latest and latest.is_running else None, + phase=phase, + phase_ends_at=ends_at if ends_at and ends_at > datetime.now(UTC) else None, ) diff --git a/backend/druks/durable/schemas.py b/backend/druks/durable/schemas.py index f549fd2a5..2b4544332 100644 --- a/backend/druks/durable/schemas.py +++ b/backend/druks/durable/schemas.py @@ -207,9 +207,10 @@ 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) - # The driving run's sandbox phase while it starts ("provisioning_vm"). The - # shell supplies the words. + # The driving run's live phase while it starts or waits. The shell supplies + # the words. A timed phase carries the time when it ends separately. phase: str | None = None + phase_ends_at: datetime | None = None class TranscriptChunk(Schema): diff --git a/backend/druks/harnesses/base.py b/backend/druks/harnesses/base.py index 45c1e068e..d4584973e 100644 --- a/backend/druks/harnesses/base.py +++ b/backend/druks/harnesses/base.py @@ -78,7 +78,8 @@ async def build_invocation(self, manifest: Manifest, **kwargs: object) -> AgentI @abstractmethod def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> object: """Turn a finished run into the structured payload (and write the - cost/output sidecars under ``artifact_dir / run_id``).""" + cost/output sidecars under ``artifact_dir / run_id``). Write any + available cost sidecar before a run failure raises.""" @classmethod def check_returncode(cls, result: HarnessRunResult) -> None: diff --git a/backend/druks/harnesses/claude.py b/backend/druks/harnesses/claude.py index 5d0352e61..bfdc45734 100644 --- a/backend/druks/harnesses/claude.py +++ b/backend/druks/harnesses/claude.py @@ -101,6 +101,7 @@ def get_acp_session( # {…overloaded_error…}"), so "api error: 5" covers 529 and every 5xx; 429 # and 401 are their own families. failure_markers = { + "monthly spend limit": exceptions.HarnessSpendLimitError, "session limit": exceptions.HarnessUsageLimitError, "usage limit": exceptions.HarnessUsageLimitError, "api error: 429": exceptions.HarnessRateLimitError, @@ -213,11 +214,12 @@ async def build_invocation( ) def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> Any: - self.check_returncode(result) - try: envelope = collapse_claude_stream(result.stdout) except exceptions.StreamJsonError as error: + # Keep a provider failure when its stream is incomplete. Only a + # successful malformed response is an output failure. + self.check_returncode(result) transcript = artifact_dir / run_id / "stdout.jsonl" raise exceptions.HarnessInvalidOutputError( f"claude wrote invalid stream-json. See {transcript}", @@ -225,6 +227,7 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> cost_usd, cost_metadata = extract_claude_cost_from_envelope(envelope) write_cost(artifact_dir / run_id, cost_usd=cost_usd, metadata=cost_metadata) + self.check_returncode(result) structured: Any = envelope.get("structured_output") if structured is None: diff --git a/backend/druks/harnesses/codex.py b/backend/druks/harnesses/codex.py index 0d5410b5e..e881be5f6 100644 --- a/backend/druks/harnesses/codex.py +++ b/backend/druks/harnesses/codex.py @@ -520,9 +520,14 @@ async def build_invocation( ) def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> Any: + call_dir = artifact_dir / run_id + cost_usd, cost_metadata = read_codex_cost_from_jsonl( + call_dir / "session.jsonl", + model=self.model_id, + ) + write_cost(call_dir, cost_usd=cost_usd, metadata=cost_metadata) self.check_returncode(result) - call_dir = artifact_dir / run_id payload = read_result_json( call_dir / "output.json", name=self.name, @@ -530,11 +535,6 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> (call_dir / "output.json").write_text( json.dumps(payload, indent=2, sort_keys=True), ) - cost_usd, cost_metadata = read_codex_cost_from_jsonl( - call_dir / "session.jsonl", - model=self.model_id, - ) - write_cost(call_dir, cost_usd=cost_usd, metadata=cost_metadata) return payload def _mcp_flags(self, servers: tuple[McpServer, ...]) -> tuple[str, ...]: diff --git a/backend/druks/harnesses/opencode.py b/backend/druks/harnesses/opencode.py index 6ae99851d..d9c829309 100644 --- a/backend/druks/harnesses/opencode.py +++ b/backend/druks/harnesses/opencode.py @@ -148,21 +148,9 @@ async def build_invocation( ) def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> Any: - if result.returncode == 124: - raise exceptions.HarnessTimeoutError( - "opencode hit the agent deadline; the session was aborted." - ) try: responses = [json.loads(line)["info"] for line in result.stdout.splitlines()] info = responses[-1] - if failure := info.get("error"): - data = failure.get("data") or {} - message = data.get("message") - detail = f"{failure['name']}: {message}" if message else failure["name"] - if failure["name"] == "APIError" and data.get("isRetryable"): - raise exceptions.HarnessOverloadedError(detail) - raise _ERROR_TYPES.get(failure["name"], exceptions.HarnessError)(detail) - structured = info["structured"] tokens = [response["tokens"] for response in responses] metadata = { "provider": self.name, @@ -174,7 +162,26 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> "cache_creation_tokens": sum(entry["cache"]["write"] for entry in tokens), } cost_usd = sum(float(response["cost"]) for response in responses) + output_dir = call_dir(artifact_dir, run_id) + write_cost(output_dir, cost_usd=cost_usd, metadata=metadata) + if result.returncode == 124: + raise exceptions.HarnessTimeoutError( + "opencode hit the agent deadline; the session was aborted." + ) + if failure := info.get("error"): + data = failure.get("data") or {} + message = data.get("message") + detail = f"{failure['name']}: {message}" if message else failure["name"] + if failure["name"] == "APIError" and data.get("isRetryable"): + raise exceptions.HarnessOverloadedError(detail) + raise _ERROR_TYPES.get(failure["name"], exceptions.HarnessError)(detail) + structured = info["structured"] + self.check_returncode(result) except (ValueError, KeyError, TypeError, AttributeError, IndexError) as error: + if result.returncode == 124: + raise exceptions.HarnessTimeoutError( + "opencode hit the agent deadline; the session was aborted." + ) from error try: self.check_returncode(result) except exceptions.HarnessError as process_error: @@ -183,8 +190,5 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> "opencode wrote no usable response.", ) from error - self.check_returncode(result) - output_dir = call_dir(artifact_dir, run_id) (output_dir / "output.json").write_text(json.dumps(structured, indent=2, sort_keys=True)) - write_cost(output_dir, cost_usd=cost_usd, metadata=metadata) return structured diff --git a/backend/druks/harnesses/pi.py b/backend/druks/harnesses/pi.py index 5d90a2880..1a779be44 100644 --- a/backend/druks/harnesses/pi.py +++ b/backend/druks/harnesses/pi.py @@ -126,7 +126,6 @@ async def build_invocation( ) def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> Any: - self.check_returncode(result) try: events = [json.loads(line) for line in result.stdout.splitlines()] # pi repeats an assistant message on turn_end and agent_end; @@ -136,6 +135,21 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> for event in events if event["type"] == "message_end" and event["message"]["role"] == "assistant" ] + usage = [message["usage"] for message in messages] + cost_usd = sum(entry["cost"]["total"] for entry in usage) + tokens = { + "input_tokens": sum(entry["input"] for entry in usage), + "output_tokens": sum(entry["output"] for entry in usage), + "cached_input_tokens": sum(entry["cacheRead"] for entry in usage), + "cache_creation_tokens": sum(entry["cacheWrite"] for entry in usage), + } + call_dir = artifact_dir / run_id + write_cost( + call_dir, + cost_usd=cost_usd, + metadata={"provider": self.name, "model": self.model, **tokens}, + ) + self.check_returncode(result) for message in messages: # A runtime provider failure exits zero, so the HTTP status in # ``errorMessage`` is what classifies it. @@ -146,24 +160,11 @@ def parse(self, result: HarnessRunResult, *, artifact_dir: Path, run_id: str) -> if code >= 500: raise exceptions.HarnessOverloadedError(detail) raise _STATUS_ERRORS.get(code, exceptions.HarnessError)(detail) - usage = [message["usage"] for message in messages] - cost_usd = sum(entry["cost"]["total"] for entry in usage) - tokens = { - "input_tokens": sum(entry["input"] for entry in usage), - "output_tokens": sum(entry["output"] for entry in usage), - "cached_input_tokens": sum(entry["cacheRead"] for entry in usage), - "cache_creation_tokens": sum(entry["cacheWrite"] for entry in usage), - } except (ValueError, KeyError, TypeError) as error: + self.check_returncode(result) raise exceptions.HarnessInvalidOutputError("pi wrote no usable stream.") from error - call_dir = artifact_dir / run_id output_path = call_dir / "output.json" payload = read_result_json(output_path, name=self.name) output_path.write_text(json.dumps(payload, indent=2, sort_keys=True)) - write_cost( - call_dir, - cost_usd=cost_usd, - metadata={"provider": self.name, "model": self.model, **tokens}, - ) return payload diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index e0f15e2a3..fc99dbe30 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -158,6 +158,7 @@ async def _dbos_cancel(workflow_id: str) -> None: mock.patch.object(Workflow, "start", classmethod(_noop)), mock.patch("druks.agents.set_run_phase", _phase_noop), mock.patch("druks.durable.reads.get_run_phase", _phase_noop), + mock.patch("druks.durable.reads.get_run_phase_ends_at", _phase_noop), mock.patch("dbos.DBOS.cancel_workflow_async", _dbos_cancel), mock.patch("dbos.DBOS.start_workflow_async", _noop), ): diff --git a/backend/tests/software_factory/test_api_work_items.py b/backend/tests/software_factory/test_api_work_items.py index b85284b68..c9934f8d7 100644 --- a/backend/tests/software_factory/test_api_work_items.py +++ b/backend/tests/software_factory/test_api_work_items.py @@ -248,11 +248,33 @@ async def test_detail_carries_the_running_phase(client: TestClient, druks_db, mo await seed_build_run(druks_db, work_item_id=item.id, state="running") async def phase(_run_id): - return "provisioning_vm" + return "waiting_for_quota" + + async def phase_ends_at(_run_id): + return datetime(2099, 8, 1, 14, 19, tzinfo=UTC) + + monkeypatch.setattr("druks.durable.reads.get_run_phase", phase) + monkeypatch.setattr("druks.durable.reads.get_run_phase_ends_at", phase_ends_at) + detail = (await client.get(f"/api/software_factory/work_item/{item.id}")).json() + assert detail["phase"] == "waiting_for_quota" + assert detail["phaseEndsAt"] == "2099-08-01T14:19:00Z" + + +async def test_detail_omits_a_phase_end_that_has_passed(client: TestClient, 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") + + async def phase(_run_id): + return "waiting_for_quota" + + async def phase_ends_at(_run_id): + return datetime(2000, 1, 1, tzinfo=UTC) monkeypatch.setattr("druks.durable.reads.get_run_phase", phase) + monkeypatch.setattr("druks.durable.reads.get_run_phase_ends_at", phase_ends_at) detail = (await client.get(f"/api/software_factory/work_item/{item.id}")).json() - assert detail["phase"] == "provisioning_vm" + + assert not detail["phaseEndsAt"] async def test_detail_has_no_phase_when_not_running(client: TestClient, druks_db): @@ -261,3 +283,4 @@ async def test_detail_has_no_phase_when_not_running(client: TestClient, druks_db detail = (await client.get(f"/api/software_factory/work_item/{item.id}")).json() assert not detail["phase"] + assert not detail["phaseEndsAt"] diff --git a/backend/tests/test_agents.py b/backend/tests/test_agents.py index 1ab243f89..ba75c5d95 100644 --- a/backend/tests/test_agents.py +++ b/backend/tests/test_agents.py @@ -867,6 +867,8 @@ async def test_body_level_quota_waits_for_the_reset_once( ) jitter = MagicMock(return_value=90.0) monkeypatch.setattr(agents.random, "uniform", jitter) + set_run_phase = AsyncMock() + monkeypatch.setattr(agents, "set_run_phase", set_run_phase) current_run._reap_run = AsyncMock() checkpoints, sleep = _inline_agent_steps @@ -876,6 +878,10 @@ async def test_body_level_quota_waits_for_the_reset_once( assert excinfo.value is failures[-1] sleep.assert_awaited_once_with(60 * 60 + 90.0) jitter.assert_called_once_with(60.0, 120.0) + set_run_phase.assert_any_await( + "waiting_for_quota", + ends_at=now + timedelta(hours=1, seconds=90), + ) current_run._reap_run.assert_awaited_once() assert [options["name"] for options in checkpoints] == [ "test.agent.dummy", diff --git a/backend/tests/test_cost_capture.py b/backend/tests/test_cost_capture.py index dae7094a4..e2e90012e 100644 --- a/backend/tests/test_cost_capture.py +++ b/backend/tests/test_cost_capture.py @@ -1,6 +1,7 @@ import json from pathlib import Path +import pytest from druks.durable import AgentCall from druks.harnesses.artifacts import ( normalize_token_usage, @@ -8,10 +9,15 @@ write_cost, ) from druks.harnesses.claude import ( + ClaudeHarness, collapse_claude_stream, extract_claude_cost_from_envelope, ) -from druks.harnesses.exceptions import StreamJsonError +from druks.harnesses.codex import CodexHarness +from druks.harnesses.exceptions import HarnessError, HarnessSpendLimitError, StreamJsonError +from druks.harnesses.opencode import OpenCodeHarness +from druks.harnesses.pi import PiHarness +from druks.sandbox.datastructures import HarnessRunResult from druks.testing import seed_call, seed_run from druks_field_notes.models import Note from druks_field_notes.workflows import Summarize @@ -52,6 +58,166 @@ def test_extract_claude_cost_returns_none_when_envelope_lacks_cost(): assert metadata is None +def test_failed_claude_call_writes_cost_before_raising(tmp_path: Path): + message = ( + "You've hit your org's monthly spend limit · ask your admin to raise it · " + "your session limit resets 2:20pm" + ) + stdout = ( + json.dumps( + { + "type": "result", + "subtype": "success", + "is_error": True, + "result": message, + "total_cost_usd": 190.25, + "usage": { + "input_tokens": 100, + "output_tokens": 20, + "cache_creation_input_tokens": 30, + "cache_read_input_tokens": 400, + }, + } + ).encode() + + b"\n" + ) + harness = ClaudeHarness(model="anthropic/claude-opus-4-7", fast_mode=False, effort=None) + + with pytest.raises(HarnessSpendLimitError): + harness.parse( + HarnessRunResult(returncode=1, stdout=stdout, stderr=b""), + artifact_dir=tmp_path, + run_id="failed-call", + ) + + cost_usd, metadata = read_cost(tmp_path / "failed-call") + assert cost_usd == 190.25 + assert metadata == { + "provider": "anthropic", + "input_tokens": 100, + "output_tokens": 20, + "cache_creation_input_tokens": 30, + "cache_read_input_tokens": 400, + } + + +def test_failed_codex_call_writes_cost_before_raising(tmp_path: Path): + call_dir = tmp_path / "failed-call" + call_dir.mkdir() + event = { + "timestamp": "2026-06-07T14:41:03.746Z", + "type": "event_msg", + "payload": { + "type": "token_count", + "info": { + "model": "gpt-5.5", + "total_token_usage": { + "input_tokens": 1_000, + "cached_input_tokens": 0, + "output_tokens": 1_000, + "reasoning_output_tokens": 0, + "total_tokens": 2_000, + }, + }, + }, + } + (call_dir / "session.jsonl").write_text(json.dumps(event) + "\n") + harness = CodexHarness(model="openai/gpt-5.5", fast_mode=False, effort=None) + + with pytest.raises(HarnessError): + harness.parse( + HarnessRunResult(returncode=1, stdout=b"", stderr=b"request failed\n"), + artifact_dir=tmp_path, + run_id="failed-call", + ) + + cost_usd, metadata = read_cost(call_dir) + assert cost_usd and cost_usd > 0 + assert metadata + assert metadata["input_tokens"] == 1_000 + assert metadata["output_tokens"] == 1_000 + + +def test_failed_pi_call_writes_cost_before_raising(tmp_path: Path): + message = { + "type": "message_end", + "message": { + "role": "assistant", + "stopReason": "error", + "errorMessage": "OpenAI API error (429): slow down", + "usage": { + "input": 100, + "output": 20, + "cacheRead": 5, + "cacheWrite": 2, + "cost": {"total": 0.75}, + }, + }, + } + harness = PiHarness(model="openai/gpt-5.5", fast_mode=False, effort=None) + + with pytest.raises(HarnessError): + harness.parse( + HarnessRunResult( + returncode=0, + stdout=(json.dumps(message) + "\n").encode(), + stderr=b"", + ), + artifact_dir=tmp_path, + run_id="failed-call", + ) + + cost_usd, metadata = read_cost(tmp_path / "failed-call") + assert cost_usd == 0.75 + assert metadata + assert metadata["input_tokens"] == 100 + assert metadata["output_tokens"] == 20 + + +def test_failed_opencode_call_writes_cost_before_raising(tmp_path: Path): + response = { + "info": { + "role": "assistant", + "providerID": "anthropic", + "modelID": "claude-sonnet-4-5", + "cost": 0.75, + "tokens": { + "input": 100, + "output": 20, + "reasoning": 3, + "cache": {"read": 5, "write": 2}, + }, + "error": { + "name": "APIError", + "data": {"message": "busy", "isRetryable": True}, + }, + }, + "parts": [], + } + harness = OpenCodeHarness( + model="anthropic/claude-sonnet-4-5", + fast_mode=False, + effort=None, + ) + + with pytest.raises(HarnessError): + harness.parse( + HarnessRunResult( + returncode=1, + stdout=(json.dumps(response) + "\n").encode(), + stderr=b"request failed\n", + ), + artifact_dir=tmp_path, + run_id="failed-call", + ) + + cost_usd, metadata = read_cost(tmp_path / "failed-call") + assert cost_usd == 0.75 + assert metadata + assert metadata["input_tokens"] == 100 + assert metadata["output_tokens"] == 20 + + def test_write_cost_and_read_cost_round_trip(tmp_path: Path): write_cost(tmp_path, cost_usd=0.5, metadata={"input_tokens": 100}) diff --git a/backend/tests/test_opencode.py b/backend/tests/test_opencode.py index 5ad560e57..85b77e466 100644 --- a/backend/tests/test_opencode.py +++ b/backend/tests/test_opencode.py @@ -4,6 +4,7 @@ from pathlib import Path import pytest +from druks.harnesses.artifacts import read_cost from druks.harnesses.base import Harness from druks.harnesses.datastructures import Manifest from druks.harnesses.exceptions import ( @@ -55,6 +56,7 @@ def _response( structured: object = None, error: dict[str, object] | None = None, cost: float = 0.0, + has_structured: bool = True, ) -> bytes: info: dict[str, object] = { "role": "assistant", @@ -70,7 +72,7 @@ def _response( } if error: info["error"] = error - else: + elif has_structured: info["structured"] = structured return json.dumps({"info": info, "parts": []}).encode() @@ -284,8 +286,21 @@ def test_parse_returns_contract_and_writes_cost_sidecars(tmp_path: Path) -> None } -@pytest.mark.parametrize("stdout", [b"", b"not-json"]) -def test_parse_rejects_empty_or_non_json_stdout(tmp_path: Path, stdout: bytes) -> None: +@pytest.mark.parametrize( + ("stdout", "expected_cost"), + [ + (b"", None), + (b"not-json", None), + ( + _response(error={"data": {"message": "missing name"}}, cost=0.25), + 0.25, + ), + (_response(cost=0.25, has_structured=False), 0.25), + ], +) +def test_parse_rejects_unusable_stdout( + tmp_path: Path, stdout: bytes, expected_cost: float | None +) -> None: with pytest.raises(HarnessInvalidOutputError): _harness().parse( HarnessRunResult(returncode=0, stdout=stdout, stderr=b""), @@ -293,6 +308,9 @@ def test_parse_rejects_empty_or_non_json_stdout(tmp_path: Path, stdout: bytes) - run_id="run-1", ) + cost_usd, _ = read_cost(tmp_path / "run-1") + assert cost_usd == expected_cost + def test_parse_maps_the_deadline_exit_to_timeout(tmp_path: Path) -> None: with pytest.raises(HarnessTimeoutError, match="deadline"): diff --git a/backend/tests/test_pi.py b/backend/tests/test_pi.py index 11a1830c1..c63c74ca2 100644 --- a/backend/tests/test_pi.py +++ b/backend/tests/test_pi.py @@ -6,6 +6,7 @@ import druks.harnesses.base import pytest from drukbox_sdk import Secret +from druks.harnesses.artifacts import read_cost from druks.harnesses.base import Harness from druks.harnesses.datastructures import Manifest from druks.harnesses.exceptions import ( @@ -256,7 +257,18 @@ def test_parse_returns_the_contract_and_sums_spend(tmp_path: Path) -> None: def test_parse_classifies_a_provider_error_by_status( tmp_path: Path, detail: str, expected: type[HarnessError] ) -> None: - stdout = _message_end("assistant", stopReason="error", errorMessage=detail) + stdout = _message_end( + "assistant", + stopReason="error", + errorMessage=detail, + usage={ + "input": 1, + "output": 1, + "cacheRead": 0, + "cacheWrite": 0, + "cost": {"total": 0.25}, + }, + ) with pytest.raises(expected) as error: _parse(stdout, tmp_path) @@ -266,7 +278,18 @@ def test_parse_classifies_a_provider_error_by_status( def test_parse_reports_a_nonzero_exit_before_reading_the_stream(tmp_path: Path) -> None: - stdout = _message_end("assistant", stopReason="error", errorMessage="OpenAI API error (401): x") + stdout = _message_end( + "assistant", + stopReason="error", + errorMessage="OpenAI API error (401): x", + usage={ + "input": 1, + "output": 1, + "cacheRead": 0, + "cacheWrite": 0, + "cost": {"total": 0.25}, + }, + ) with pytest.raises(HarnessError, match="pi exited with 1.*extension failed") as error: _parse(stdout, tmp_path, returncode=1, stderr=b"extension failed\n") @@ -279,9 +302,33 @@ def test_parse_treats_a_missing_result_file_as_invalid_output(tmp_path: Path) -> _parse(b'{"type":"agent_settled"}', tmp_path) -def test_parse_treats_a_broken_stream_as_invalid_output(tmp_path: Path) -> None: +@pytest.mark.parametrize( + ("stdout", "expected_cost"), + [ + (b"not json", None), + ( + _message_end( + "assistant", + usage={ + "input": 1, + "output": 1, + "cacheRead": 0, + "cacheWrite": 0, + "cost": {"total": 0.25}, + }, + ), + 0.25, + ), + ], +) +def test_parse_treats_a_broken_stream_as_invalid_output( + tmp_path: Path, stdout: bytes, expected_cost: float | None +) -> None: with pytest.raises(HarnessInvalidOutputError, match="no usable stream"): - _parse(b"not json", tmp_path) + _parse(stdout, tmp_path) + + cost_usd, _ = read_cost(tmp_path / "run-1") + assert cost_usd == expected_cost async def test_a_pasted_key_becomes_the_catalog_entry(client, druks_db) -> None: diff --git a/backend/tests/test_sandboxed_harness.py b/backend/tests/test_sandboxed_harness.py index 76c14432f..54253366d 100644 --- a/backend/tests/test_sandboxed_harness.py +++ b/backend/tests/test_sandboxed_harness.py @@ -23,6 +23,7 @@ HarnessOverloadedError, HarnessRateLimitError, HarnessSandboxError, + HarnessSpendLimitError, HarnessTimeoutError, HarnessUsageLimitError, Retry, @@ -529,6 +530,14 @@ def _claude_result_event(text: str) -> bytes: HarnessUsageLimitError, "usage_limit", ), + ( + ClaudeHarness, + _claude_result_event( + "You've hit your monthly spend limit · your session limit resets 4:20pm" + ), + HarnessSpendLimitError, + "spend_limit", + ), ( ClaudeHarness, _claude_result_event( diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts index c13216c80..2a8442625 100644 --- a/frontend/src/api/types.ts +++ b/frontend/src/api/types.ts @@ -178,14 +178,15 @@ 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 -// driving run's sandbox phase while it starts. +// driving run's live phase while it starts or waits. export interface SubjectResponse { summary: S status: SubjectStatus timeline: RunSummary[] - // The driving run's sandbox phase while it starts ("provisioning_vm"). The shell - // supplies the words. + // The driving run's live phase while it starts or waits. The shell supplies + // the words. A timed phase carries the time when it ends separately. phase?: string | null + phaseEndsAt?: string | null } export interface ArtifactFile { diff --git a/frontend/src/apps/software_factory/WorkItemPage.tsx b/frontend/src/apps/software_factory/WorkItemPage.tsx index 7036ef21f..9a8be71b6 100644 --- a/frontend/src/apps/software_factory/WorkItemPage.tsx +++ b/frontend/src/apps/software_factory/WorkItemPage.tsx @@ -28,6 +28,7 @@ import { agentCallPath, workItemPath } from './slug' import { useRawLocation } from '../../lib/useRawLocation' import { useCanonicalPath } from '../../lib/useCanonicalPath' import { useTicker } from '../../lib/useTicker' +import { useTimezone } from '../../lib/preferences' interface Props { workItemId: number @@ -195,6 +196,7 @@ function WorkItemView({ data }: { data: WorkItemDetail }) { setSelected(id)} /> @@ -313,11 +315,13 @@ function InfoPanel({ function TimelinePanel({ runs, phase, + phaseEndsAt, selection, onSelect, }: { runs: RunSummary[] phase?: string | null + phaseEndsAt?: string | null selection: Selection | null onSelect: (id: string) => void }) { @@ -335,6 +339,7 @@ function TimelinePanel({ key={chain[0].id} chain={chain} phase={phase} + phaseEndsAt={phaseEndsAt} selection={selection} onSelect={onSelect} /> @@ -357,11 +362,13 @@ function selectOnKey(onSelect: () => void) { function ChainRow({ chain, phase, + phaseEndsAt, selection, onSelect, }: { chain: RetryChain phase?: string | null + phaseEndsAt?: string | null selection: Selection | null onSelect: (id: string) => void }) { @@ -397,7 +404,14 @@ function ChainRow({ {first.accountUsername} - {collapseCalls && } + {collapseCalls && ( + + )}
{elapsed > 0 ? dur(elapsed) : '–'} @@ -411,6 +425,7 @@ function ChainRow({ run={run} retry={index} phase={phase} + phaseEndsAt={phaseEndsAt} selection={selection} onSelect={onSelect} /> @@ -426,12 +441,14 @@ function RunSegment({ run, retry, phase, + phaseEndsAt, selection, onSelect, }: { run: RunSummary retry: number phase?: string | null + phaseEndsAt?: string | null selection: Selection | null onSelect: (id: string) => void }) { @@ -478,7 +495,12 @@ function RunSegment({ onClick={selectRun} onKeyDown={selectOnKey(selectRun)} > - +
)} @@ -489,16 +511,19 @@ function RunSegment({ function RunStatus({ run, phase, + phaseEndsAt, collapsed, }: { run: RunSummary phase?: string | null + phaseEndsAt?: string | null collapsed: boolean }) { + const timezone = useTimezone() return ( - {runSubLine(run, phase, collapsed)} + {runSubLine(run, phase, collapsed, phaseEndsAt, timezone)} {!isRunning(run) && · {relTime(secondsSince(run.updatedAt))}} ) @@ -742,6 +767,7 @@ function TranscriptBody({ run: RunSummary call: AgentCallSummary | null }) { + const timezone = useTimezone() const isLive = call?.status === 'running' // Running but no agent call yet → the sandbox spin-up window. The run's sandbox @@ -753,7 +779,9 @@ function TranscriptBody({
◍
-
{phaseLine(data.phase) ?? 'Starting up…'}
+
+ {phaseLine(data.phase, data.phaseEndsAt, timezone) ?? 'Starting up…'} +
no agent call yet — the transcript begins once the agent starts
diff --git a/frontend/src/apps/software_factory/statusLine.test.ts b/frontend/src/apps/software_factory/statusLine.test.ts index b4e2f8a02..f79fa6045 100644 --- a/frontend/src/apps/software_factory/statusLine.test.ts +++ b/frontend/src/apps/software_factory/statusLine.test.ts @@ -119,6 +119,12 @@ describe('runSubLine', () => { expect(runSubLine(run(), 'agent_running', true)).toBe('running') }) + it('names a quota wait and its end in the account timezone', () => { + expect(runSubLine(run(), 'waiting_for_quota', true, '2026-08-01T14:19:00Z', 'UTC')).toBe( + 'Waiting for the usage window, tries again at 14:19', + ) + }) + it('falls back to the state when nothing is running yet', () => { expect(runSubLine(run({ state: 'scheduled' }), null, true)).toBe('scheduled') }) diff --git a/frontend/src/apps/software_factory/statusLine.ts b/frontend/src/apps/software_factory/statusLine.ts index 8aa8f440c..caea84d2e 100644 --- a/frontend/src/apps/software_factory/statusLine.ts +++ b/frontend/src/apps/software_factory/statusLine.ts @@ -57,6 +57,8 @@ export function runSubLine( run: RunSummary, phase: string | null | undefined, collapsed: boolean, + phaseEndsAt?: string | null, + timezone: string = 'UTC', ): string { if (run.state === 'failed' && run.failure) { return (run.failure.split('—')[0] ?? run.failure).trim() @@ -67,7 +69,7 @@ export function runSubLine( if (run.state === 'running') { const step = collapsed && run.agentCalls.find((call) => call.status === 'running') if (step) return step.label - const line = phaseLine(phase) + const line = phaseLine(phase, phaseEndsAt, timezone) if (line) return line } return STATE_LABEL[run.state] ?? run.state diff --git a/frontend/src/lib/phase.ts b/frontend/src/lib/phase.ts index 4b85433b5..add6b5a7f 100644 --- a/frontend/src/lib/phase.ts +++ b/frontend/src/lib/phase.ts @@ -1,11 +1,22 @@ -// The sandbox phases a run pushes before its first agent call, in words. A -// running agent call names itself, so the later phase maps to nothing. +import { clockTime } from './format' + +// The transient phases a run pushes while it starts or waits, in words. A +// running agent call names itself, so that phase maps to nothing. const PHASE_LINES: Record = { provisioning_vm: 'Provisioning sandbox VM…', queued_for_vm: 'Waiting for a sandbox slot…', sandbox_building: 'Building sandbox…', + waiting_for_quota: 'Waiting for the usage window…', } -export function phaseLine(phase: string | null | undefined): string | null { - return phase ? (PHASE_LINES[phase] ?? null) : null +export function phaseLine( + phase: string | null | undefined, + endsAt?: string | null, + timezone: string = 'UTC', +): string | null { + if (!phase) return null + if (phase === 'waiting_for_quota' && endsAt) { + return `Waiting for the usage window, tries again at ${clockTime(endsAt, timezone)}` + } + return PHASE_LINES[phase] ?? null } diff --git a/frontend/src/pages/SubjectPage.tsx b/frontend/src/pages/SubjectPage.tsx index e27f64829..a37ea7a33 100644 --- a/frontend/src/pages/SubjectPage.tsx +++ b/frontend/src/pages/SubjectPage.tsx @@ -16,6 +16,7 @@ import { RunTranscript } from '../components/RunTranscript' import { StatusGlyph } from '../components/StatusGlyph' import { httpUrl, relTimeFromIso } from '../lib/format' import { phaseLine } from '../lib/phase' +import { useTimezone } from '../lib/preferences' import { summaryEntries } from '../lib/summary' const isActiveRun = (run: RunSummary) => @@ -58,6 +59,7 @@ function SubjectDetail({ subjectId: string search: string }) { + const timezone = useTimezone() const target = new URLSearchParams(search) const selectedRun = target.get('run') const parkedAt = target.get('parkedAt') ?? undefined @@ -99,7 +101,7 @@ function SubjectDetail({ const data = query.data! const runs = [...data.timeline].reverse() - const now = phaseLine(data.phase) + const now = phaseLine(data.phase, data.phaseEndsAt, timezone) const crumb = (