Add oa run --usage PATH so scripts can meter spend

CHANGELOG.md

@@ -29,6 +29,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Sandbox domain rules cover ports and MCP endpoints** — `http.allow_domains` entries may pin `host:port` while bare hosts retain backwards-compatible any-port semantics. Statically configured MCP endpoints are preflighted across the selected task and direct dependencies before discovery or model execution. Malformed allowlist entries now fail validation. (#104)
 - **Declared remote delegated-spec URLs respect the delegating task's sandbox** — `http://`, `https://`, and resolved `oa://` destinations are checked against that task's effective `sandbox.http.allow_domains` before the initial request. Redirect destinations remain tracked in #114; cross-document sandbox inheritance remains tracked in #110. (#112)
 
+### Added
+- **`oa run --usage PATH`** — writes a usage JSON file (leaf task, `depends_on` chain, and rolled-up `total`) without changing stdout, so `--quiet` scripts can meter spend. Leaf-only would under-count chained tasks; `estimated_cost_usd` is included on `total` only when every contributing block has a cost. (#106)
+
 ### Fixed
 - **npm CLI accepts a bare spec path** — `oa validate <spec.yaml>` and `oa run <spec.yaml>` now work without `--spec` in the npm runtime, matching the Python CLI (1.6.0). Same guardrails: `--spec` unchanged, bare path + `--spec` together is an explicit error, and a non-YAML bare argument gets a clear error naming the valid forms. First Jest tests land with this (`npm/tests/`), and both CI and the npm publish workflow now run them. (#100)
 

docs/REFERENCE.md

@@ -358,8 +358,29 @@ so you can track and budget spend without a separate accounting layer:
   growing history, so the sum reflects what is actually billed).
 
 `oa run` (without `--quiet`) shows a compact `<total> tok · ~$<cost>` summary in
-the result panel; `--quiet` emits only the task output, so read `usage` from the
-full envelope returned by the Python API when scripting.
+the result panel. `--quiet` still emits only the task output on stdout (clean
+for `| jq`); pass `--usage PATH` to write a JSON file with the leaf `usage`,
+any `depends_on` chain, and a rolled-up `total` so a script can meter spend
+without under-counting chained tasks:
+
+```bash
+oa run --spec .agents/example.yaml --task greet \
+  --input '{"name":"Alice"}' --quiet --usage /tmp/usage.json
+```
+
+```json
+{
+  "task": "summarize",
+  "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+  "chain": {
+    "extract": {
+      "task": "extract",
+      "usage": {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30}
+    }
+  },
+  "total": {"prompt_tokens": 30, "completion_tokens": 15, "total_tokens": 45}
+}
+```
 
 #### Overriding the cost rate
 

oas_cli/main.py

@@ -29,6 +29,7 @@
     print_result_panel,
     print_run_header,
 )
+from .usage import report_from_envelope
 
 app = typer.Typer(help="Open Agent (OA) CLI")
 console = Console()
@@ -635,6 +636,14 @@ def run(
         "-q",
         help="No banner; print only the task output JSON to stdout (clean for | jq)",
     ),
+    usage_path: Path | None = typer.Option(
+        None,
+        "--usage",
+        help=(
+            "Write usage JSON (leaf, depends_on chain, and rolled-up total) "
+            "to this path. stdout is unchanged."
+        ),
+    ),
 ):
     """Run a single task directly from an Open Agent Spec file.
 
@@ -752,6 +761,13 @@ def run(
 
         elapsed = time.monotonic() - t0
 
+        if usage_path is not None:
+            usage_path.parent.mkdir(parents=True, exist_ok=True)
+            usage_path.write_text(
+                json.dumps(report_from_envelope(result), indent=2) + "\n",
+                encoding="utf-8",
+            )
+
         if quiet:
             # Print only the agent's output for clean piping.
             # dict/list → pretty JSON; plain string → written directly so no

oas_cli/ui.py

@@ -310,7 +310,8 @@ def print_help_panel(console: Console, version: str = "") -> None:
         f"  [{_C_BRAND}]oa test[/]   [white]agent.test.yaml[/]\n"
         f"  [{_C_BRAND}]oa update[/] [white]--spec path.yaml --output dir/[/]\n"
         "\n"
-        f"  [dim]--quiet / -q[/]  [dim]JSON-only output for pipes and CI[/]"
+        f"  [dim]--quiet / -q[/]  [dim]JSON-only output for pipes and CI[/]\n"
+        f"  [dim]--usage PATH[/]  [dim]write usage JSON (leaf, chain, total)[/]"
     )
     console.print(
         Panel.fit(

oas_cli/usage.py

@@ -34,6 +34,7 @@
 
 import json
 import os
+from typing import Any
 
 CanonicalUsage = dict[str, int]
 
@@ -125,6 +126,66 @@ def _canonical(
 }
 
 
+def report_from_envelope(result: dict[str, Any]) -> dict[str, Any]:
+    """Build a metering report from a task-result envelope.
+
+    Mirrors the envelope's usage tree (the leaf task plus any ``chain`` of
+    ``depends_on`` envelopes) and includes a rolled-up ``total`` so a script
+    metering a chained task cannot under-count by reading the leaf block
+    alone. ``estimated_cost_usd`` is included on ``total`` only when every
+    contributing block has a numeric cost — a silently incomplete dollar
+    figure is worse than none.
+    """
+    report = _usage_node(result)
+    report["total"] = _sum_usage(_collect_usage(result))
+    return report
+
+
+def _usage_node(result: dict[str, Any]) -> dict[str, Any]:
+    """Leaf of the usage tree: task name, usage, and nested chain if present."""
+    node: dict[str, Any] = {
+        "task": result.get("task"),
+        "usage": result.get("usage"),
+    }
+    chain = result.get("chain")
+    if isinstance(chain, dict) and chain:
+        node["chain"] = {
+            name: _usage_node(dep) if isinstance(dep, dict) else dep
+            for name, dep in chain.items()
+        }
+    return node
+
+
+def _collect_usage(result: dict[str, Any]) -> list[dict[str, Any]]:
+    """Walk the envelope (leaf then chain) collecting non-null usage dicts."""
+    blocks: list[dict[str, Any]] = []
+    usage = result.get("usage")
+    if isinstance(usage, dict):
+        blocks.append(usage)
+    chain = result.get("chain")
+    if isinstance(chain, dict):
+        for dep in chain.values():
+            if isinstance(dep, dict):
+                blocks.extend(_collect_usage(dep))
+    return blocks
+
+
+def _sum_usage(blocks: list[dict[str, Any]]) -> dict[str, Any] | None:
+    """Sum token counts across *blocks*. Cost is omitted unless every block has one."""
+    if not blocks:
+        return None
+    total: dict[str, Any] = {
+        "prompt_tokens": sum(int(b.get("prompt_tokens") or 0) for b in blocks),
+        "completion_tokens": sum(int(b.get("completion_tokens") or 0) for b in blocks),
+        "total_tokens": sum(int(b.get("total_tokens") or 0) for b in blocks),
+    }
+    costs = [b.get("estimated_cost_usd") for b in blocks]
+    numeric = [c for c in costs if isinstance(c, (int, float))]
+    if numeric and len(numeric) == len(costs):
+        total["estimated_cost_usd"] = round(sum(float(c) for c in numeric), 6)
+    return total
+
+
 def estimate_cost_usd(
     model: str | None,
     usage: CanonicalUsage | None,

tests/test_main.py

@@ -231,3 +231,170 @@ def fake_invoke(system: str, user: str, config: dict, history=None) -> str:
     assert result.exit_code == 0, result.output
     parsed = json.loads(result.output)
     assert parsed.get("response") == "Hello Alice!"
+
+
+# ---------------------------------------------------------------------------
+# oa run --usage PATH
+# ---------------------------------------------------------------------------
+
+_CHAIN_SPEC = """\
+open_agent_spec: "1.5.0"
+
+agent:
+  name: test-agent
+  description: test agent
+
+intelligence:
+  type: llm
+  engine: openai
+  model: gpt-4o
+
+tasks:
+  extract:
+    description: extract facts
+    output:
+      type: object
+      properties:
+        facts: { type: string }
+      required: [facts]
+    prompts:
+      user: "extract facts"
+  summarize:
+    description: summarize facts
+    depends_on: [extract]
+    output:
+      type: object
+      properties:
+        summary: { type: string }
+      required: [summary]
+
+prompts:
+  system: "you summarize"
+  user: "{{ facts }}"
+"""
+
+
+def test_run_usage_flag_writes_file_and_leaves_quiet_stdout_clean(tmp_path):
+    """--usage writes the report to a file; --quiet stdout stays output-only."""
+    spec_file = _write_spec(tmp_path)
+    usage_file = tmp_path / "usage.json"
+
+    def fake_invoke(system: str, user: str, config: dict, history=None) -> str:
+        from oas_cli.providers.registry import record_usage
+
+        record_usage(
+            {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            "gpt-4o",
+        )
+        return '{"response": "Hello Alice!"}'
+
+    with patch("oas_cli.runner.invoke_intelligence", fake_invoke):
+        result = runner.invoke(
+            app,
+            [
+                "run",
+                "--spec",
+                str(spec_file),
+                "--task",
+                "greet",
+                "--input",
+                '{"name": "Alice"}',
+                "--quiet",
+                "--usage",
+                str(usage_file),
+            ],
+        )
+
+    assert result.exit_code == 0, result.output
+    parsed = json.loads(result.output)
+    assert parsed == {"response": "Hello Alice!"}
+    assert "usage" not in parsed
+    report = json.loads(usage_file.read_text(encoding="utf-8"))
+    assert report["task"] == "greet"
+    assert report["usage"]["total_tokens"] == 15
+    assert report["total"]["total_tokens"] == 15
+    assert "chain" not in report
+
+
+def test_run_usage_flag_rolls_up_depends_on_chain(tmp_path):
+    """--usage total includes dependency spend, not just the leaf task."""
+    spec_file = tmp_path / "agent.yaml"
+    spec_file.write_text(_CHAIN_SPEC)
+    usage_file = tmp_path / "usage.json"
+
+    def fake_invoke(system: str, user: str, config: dict, history=None) -> str:
+        from oas_cli.providers.registry import record_usage
+
+        if "extract facts" in user:
+            record_usage(
+                {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30},
+                "gpt-4o",
+            )
+            return '{"facts": "the sky is blue"}'
+        record_usage(
+            {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            "gpt-4o",
+        )
+        return '{"summary": "sky=blue"}'
+
+    with patch("oas_cli.runner.invoke_intelligence", fake_invoke):
+        result = runner.invoke(
+            app,
+            [
+                "run",
+                "--spec",
+                str(spec_file),
+                "--task",
+                "summarize",
+                "--quiet",
+                "--usage",
+                str(usage_file),
+            ],
+        )
+
+    assert result.exit_code == 0, result.output
+    parsed = json.loads(result.output)
+    assert parsed == {"summary": "sky=blue"}
+    report = json.loads(usage_file.read_text(encoding="utf-8"))
+    assert report["task"] == "summarize"
+    assert report["usage"]["total_tokens"] == 15
+    assert report["chain"]["extract"]["usage"]["total_tokens"] == 30
+    assert report["total"]["total_tokens"] == 45
+    assert report["total"]["total_tokens"] != report["usage"]["total_tokens"]
+    assert "estimated_cost_usd" in report["total"]
+
+
+def test_run_usage_flag_works_without_quiet(tmp_path):
+    """--usage writes the same report in non-quiet mode."""
+    spec_file = _write_spec(tmp_path)
+    usage_file = tmp_path / "usage.json"
+
+    def fake_invoke(system: str, user: str, config: dict, history=None) -> str:
+        from oas_cli.providers.registry import record_usage
+
+        record_usage(
+            {"prompt_tokens": 8, "completion_tokens": 2, "total_tokens": 10},
+            "gpt-4o",
+        )
+        return '{"response": "hi"}'
+
+    with patch("oas_cli.runner.invoke_intelligence", fake_invoke):
+        result = runner.invoke(
+            app,
+            [
+                "run",
+                "--spec",
+                str(spec_file),
+                "--task",
+                "greet",
+                "--input",
+                '{"name": "Alice"}',
+                "--usage",
+                str(usage_file),
+            ],
+        )
+
+    assert result.exit_code == 0, result.output
+    report = json.loads(usage_file.read_text(encoding="utf-8"))
+    assert report["usage"]["total_tokens"] == 10
+    assert report["total"]["total_tokens"] == 10