diff --git a/adapters/python/claude/README.md b/adapters/python/claude/README.md index f4d435672..a6b8ac271 100644 --- a/adapters/python/claude/README.md +++ b/adapters/python/claude/README.md @@ -149,8 +149,8 @@ output tokens. Harbor accounting reads these counters before any ATIF trajectory. With the input counters present, Harbor sets `n_input_tokens` to input plus cache reads plus cache writes and `n_cache_tokens` to cache reads, from the same main-loop -scope as `n_output_tokens`. A Relay trajectory then fills only the counts the -result leaves unknown. +scope as `n_output_tokens`. Counts the result leaves unknown stay unknown in +Harbor; a Relay trajectory's totals are recorded separately and never fill them. ## Relay Observability diff --git a/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/README.md b/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/README.md index 14f33030b..dbfc79ea2 100644 --- a/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/README.md +++ b/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/README.md @@ -61,7 +61,9 @@ The task runner validates telemetry after the one-shot NeMo Fabric runtime shuts ## Invocation Accounting -The integration projects normalized `RunResult.usage` into Harbor's `AgentContext` without requiring NeMo Relay or ATIF. Harbor input counts include cache: inclusive NeMo Fabric input is copied unchanged, while exclusive input is combined with cache only when both counts are known. If input semantics are unknown, Harbor input remains unknown unless the adapter explicitly reports zero cache tokens. Legacy results remain valid, but their input count needs explicit cache semantics or an ATIF fallback to populate Harbor's inclusive count; the original values remain in result metadata. Unknown counts and costs remain `None`; estimates are not promoted to reported cost. ATIF metrics fill only missing fields and are never added to normalized usage. The same collection path applies to unsuccessful results that contain usage. +The integration projects normalized `RunResult.usage` into Harbor's `AgentContext` without requiring NeMo Relay or ATIF. Harbor input counts include cache: inclusive NeMo Fabric input is copied unchanged, while exclusive input is combined with cache only when both counts are known. If input semantics are unknown, Harbor input remains unknown unless the adapter explicitly reports zero cache tokens. Legacy results remain valid, but their input count needs explicit cache semantics to populate Harbor's inclusive count; the original values remain in result metadata. Unknown counts and costs remain `None` and reported zeros remain zero; estimates are not promoted to reported cost. When the runner exits without a NeMo Fabric result, the integration sets all four Harbor accounting fields to `None`. ATIF `final_metrics` never fill these fields, because their scope and cache conventions are not verified against the invocation. They are kept under `metadata.fabric.atif_final_metrics` with `source: atif` and `coverage: unverified`. The same collection path applies to unsuccessful results that contain usage. + +**Reporting limitation:** Native Harbor trial and job totals sum available values and skip unknown fields. A displayed total can therefore omit usage or cost from steps or trials with unknown values. Do not interpret these aggregates as complete benchmark usage or expenditure when any contributing value is unknown. ## Host Admission And Task Execution diff --git a/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/fabric_agent.py b/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/fabric_agent.py index 2c7e25a95..bc274efb8 100644 --- a/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/fabric_agent.py +++ b/sdk/python/nemo-fabric-runtime/src/nemo_fabric/integrations/harbor/fabric_agent.py @@ -65,6 +65,12 @@ HARBOR_ARTIFACT_ROOT = "/logs/agent/fabric-artifacts" HARBOR_DEFAULT_WORKSPACE = "/testbed" _DETAILED_TOKEN_USAGE_V1_KEY = "nemo_fabric.detailed_token_usage.v1" +# Evidence the bridge records beside the result, kept when result metadata is replaced. +_BRIDGE_EVIDENCE_KEYS = ( + "atif_final_metrics", + "harbor_atif_validation", + "telemetry_validation", +) try: from harbor.agents.base import BaseAgent @@ -614,6 +620,8 @@ def populate_context_post_run(self, context: AgentContext) -> None: if self._result_path is not None: populate_context_from_result(context, self._result_path) + else: + _clear_accounting(context) populate_context_from_trajectory(context, self.logs_dir / "trajectory.json") populate_context_from_telemetry_summary( context, @@ -884,26 +892,33 @@ def populate_context_from_result( document = json.loads(path.read_text(encoding="utf-8")) if isinstance(document, dict) and "runner_error" in document: failure = FabricRunnerFailure.model_validate(document) - if context.metadata is None: - context.metadata = {} - context.metadata["fabric"] = failure.model_dump() + _replace_result_metadata(context, failure.model_dump()) + _clear_accounting(context) return failure result = RunResult.from_mapping(document) mapping = result.to_mapping() - if context.metadata is None: - context.metadata = {} - context.metadata["fabric"] = { - "status": mapping["status"], - "runtime_id": mapping["runtime_id"], - "invocation_id": mapping["invocation_id"], - "request_id": mapping["request_id"], - "harness": mapping["harness"], - "adapter_id": mapping.get("adapter_id"), - "artifacts": mapping["artifacts"], - "telemetry": mapping["telemetry"], - "error": mapping.get("error"), - "usage": mapping.get("usage"), - } + _replace_result_metadata( + context, + { + "status": mapping["status"], + "runtime_id": mapping["runtime_id"], + "invocation_id": mapping["invocation_id"], + "request_id": mapping["request_id"], + "harness": mapping["harness"], + "adapter_id": mapping.get("adapter_id"), + "artifacts": mapping["artifacts"], + "telemetry": mapping["telemetry"], + "error": mapping.get("error"), + "usage": mapping.get("usage"), + }, + ) + # Fabric results supply Harbor's accounting fields; an unknown dimension + # clears any value an earlier pass left in the context. ATIF metrics stay + # separate as evidence with unverified coverage. + input_tokens: int | None = None + cached_input_tokens: int | None = None + output_tokens: int | None = None + cost_usd: float | None = None if result.usage is not None: usage = result.usage detailed_usage = _detailed_token_usage(usage) @@ -959,19 +974,39 @@ def populate_context_from_result( ) else: output_tokens = usage.get("output_tokens") - if input_tokens is not None: - context.n_input_tokens = input_tokens - if cached_input_tokens is not None: - context.n_cache_tokens = cached_input_tokens - if output_tokens is not None: - context.n_output_tokens = output_tokens - if usage.get("cost_usd") is not None: - context.cost_usd = usage["cost_usd"] + cost_usd = usage.get("cost_usd") + context.n_input_tokens = input_tokens + context.n_cache_tokens = cached_input_tokens + context.n_output_tokens = output_tokens + context.cost_usd = cost_usd return result +def _replace_result_metadata(context: AgentContext, fields: dict[str, Any]) -> None: + """Replace one pass's result fields while keeping the bridge's own evidence.""" + + if context.metadata is None: + context.metadata = {} + previous = context.metadata.get("fabric") + evidence = ( + {key: previous[key] for key in _BRIDGE_EVIDENCE_KEYS if key in previous} + if isinstance(previous, dict) + else {} + ) + context.metadata["fabric"] = {**fields, **evidence} + + +def _clear_accounting(context: AgentContext) -> None: + """Without a Fabric result, no source can vouch for Harbor's accounting fields.""" + + context.n_input_tokens = None + context.n_cache_tokens = None + context.n_output_tokens = None + context.cost_usd = None + + def populate_context_from_trajectory(context: AgentContext, path: Path) -> None: - """Validate canonical ATIF with Harbor and backfill usage fields when present.""" + """Validate canonical ATIF with Harbor and keep its totals as separate evidence.""" if not path.is_file(): return @@ -986,15 +1021,23 @@ def populate_context_from_trajectory(context: AgentContext, path: Path) -> None: metrics = trajectory.final_metrics if metrics is None: return - # These are two views of one invocation, not additive accounting sources. - if context.n_input_tokens is None: - context.n_input_tokens = metrics.total_prompt_tokens - if context.n_cache_tokens is None: - context.n_cache_tokens = metrics.total_cached_tokens - if context.n_output_tokens is None: - context.n_output_tokens = metrics.total_completion_tokens - if context.cost_usd is None: - context.cost_usd = metrics.total_cost_usd + # ATIF totals can cover a different scope (for example subagents) and cache + # convention than the invocation, so they never fill Harbor's accounting. + _record_atif_final_metrics(context, metrics.model_dump(mode="json")) + + +def _record_atif_final_metrics( + context: AgentContext, final_metrics: dict[str, Any] +) -> None: + if context.metadata is None: + context.metadata = {} + fabric = context.metadata.setdefault("fabric", {}) + if isinstance(fabric, dict): + fabric["atif_final_metrics"] = { + "source": "atif", + "coverage": "unverified", + "final_metrics": final_metrics, + } def _record_host_atif_validation( diff --git a/tests/integrations/test_harbor_runner.py b/tests/integrations/test_harbor_runner.py index c2d10221f..b98431af3 100644 --- a/tests/integrations/test_harbor_runner.py +++ b/tests/integrations/test_harbor_runner.py @@ -707,10 +707,17 @@ async def download_file(self, _source, destination): ) assert context.metadata["fabric"]["status"] == "succeeded" - assert context.n_input_tokens == 12 - assert context.n_cache_tokens == 3 - assert context.n_output_tokens == 4 - assert context.cost_usd == 0.25 + atif_metrics = context.metadata["fabric"]["atif_final_metrics"] + assert atif_metrics["source"] == "atif" + assert atif_metrics["coverage"] == "unverified" + assert atif_metrics["final_metrics"]["total_prompt_tokens"] == 12 + assert atif_metrics["final_metrics"]["total_cached_tokens"] == 3 + assert atif_metrics["final_metrics"]["total_completion_tokens"] == 4 + assert atif_metrics["final_metrics"]["total_cost_usd"] == 0.25 + assert context.n_input_tokens is None + assert context.n_cache_tokens is None + assert context.n_output_tokens is None + assert context.cost_usd is None def test_harbor_023_options_schema_and_preflight(): diff --git a/tests/python/test_harbor_accounting.py b/tests/python/test_harbor_accounting.py index 238c55765..e549a5843 100644 --- a/tests/python/test_harbor_accounting.py +++ b/tests/python/test_harbor_accounting.py @@ -248,12 +248,65 @@ def test_detailed_usage_is_authoritative_per_dimension( } +ATIF_FINAL_METRICS = { + "total_prompt_tokens": 12, + "total_completion_tokens": 4, + "total_cached_tokens": 3, + "total_cost_usd": 0.25, + "total_steps": None, + "extra": None, +} + + +@pytest.fixture(name="trajectory_path") +def trajectory_path_fixture(tmp_path): + path = tmp_path / "trajectory.json" + path.write_text( + json.dumps( + { + "schema_version": "ATIF-v1.7", + "session_id": "runtime-1", + "agent": {"name": "fabric", "version": "0.1.0"}, + "steps": [{"step_id": 1, "source": "agent", "message": "done"}], + "final_metrics": { + "total_prompt_tokens": 12, + "total_cached_tokens": 3, + "total_completion_tokens": 4, + "total_cost_usd": 0.25, + }, + } + ) + ) + return path + + +@pytest.fixture(name="failure_path") +def failure_path_fixture(tmp_path): + path = tmp_path / "failure.json" + path.write_text( + json.dumps( + { + "runner_error": { + "stage": "start", + "code": "configuration_failed", + "message": "startup failure", + "retryable": False, + } + } + ) + ) + return path + + @pytest.mark.parametrize( ("usage", "expected"), [ - ({"metadata": {"estimated_cost_usd": 9}}, (12, 3, 4, 0.25)), - ({"input_tokens": 20, "cached_input_tokens": 5}, (12, 5, 4, 0.25)), - ({"input_tokens": 20, "input_tokens_include_cache": False}, (12, 3, 4, 0.25)), + ({"metadata": {"estimated_cost_usd": 9}}, (None, None, None, None)), + ({"input_tokens": 20, "cached_input_tokens": 5}, (None, 5, None, None)), + ( + {"input_tokens": 20, "input_tokens_include_cache": False}, + (None, None, None, None), + ), ( { "input_tokens": 0, @@ -264,9 +317,24 @@ def test_detailed_usage_is_authoritative_per_dimension( }, (0, 0, 0, 0), ), + pytest.param(None, (None, None, None, None), id="usage-null"), + pytest.param( + { + "input_tokens": 20, + "cached_input_tokens": 5, + "input_tokens_include_cache": True, + "output_tokens": 7, + "cost_usd": 1.0, + "metadata": { + "nemo_fabric.detailed_token_usage.v1": {"reasoning_tokens": 4} + }, + }, + (20, 5, None, 1.0), + id="detailed-usage-makes-output-unavailable", + ), ], ) -def test_unknown_usage_preserves_existing_context(result_path, usage, expected): +def test_fabric_usage_replaces_existing_context(result_path, usage, expected): from harbor.models.agent.context import AgentContext from nemo_fabric.integrations.harbor.fabric_agent import ( populate_context_from_result, @@ -275,6 +343,7 @@ def test_unknown_usage_preserves_existing_context(result_path, usage, expected): document = json.loads(result_path.read_text()) document["usage"] = usage result_path.write_text(json.dumps(document)) + # Values an earlier ATIF fallback could have left behind. context = AgentContext( n_input_tokens=12, n_cache_tokens=3, n_output_tokens=4, cost_usd=0.25 ) @@ -290,12 +359,13 @@ def test_unknown_usage_preserves_existing_context(result_path, usage, expected): ) == expected +@pytest.mark.parametrize("trajectory_first", [False, True]) @pytest.mark.parametrize( ("usage", "expected"), [ ( {"input_tokens": 0, "input_tokens_include_cache": True, "cost_usd": 0}, - (0, 3, 4, 0), + (0, None, None, 0), ), ( { @@ -307,7 +377,7 @@ def test_unknown_usage_preserves_existing_context(result_path, usage, expected): }, (20, 5, 8, 1.25), ), - (None, (12, 3, 4, 0.25)), + (None, (None, None, None, None)), pytest.param( { "input_tokens": 5, @@ -323,8 +393,8 @@ def test_unknown_usage_preserves_existing_context(result_path, usage, expected): } }, }, - (3405, 3000, 90, 0.25), - id="complete-detailed-input-takes-input-and-cache-from-result", + (3405, 3000, 90, None), + id="complete-detailed-input-leaves-unreported-cost-unknown", ), pytest.param( { @@ -337,12 +407,14 @@ def test_unknown_usage_preserves_existing_context(result_path, usage, expected): } }, }, - (12, 3, 90, 0.25), - id="output-only-detailed-takes-input-and-cache-from-atif", + (None, None, 90, None), + id="output-only-detailed-leaves-input-and-cache-unknown", ), ], ) -def test_atif_only_fills_missing_accounting(result_path, tmp_path, usage, expected): +def test_atif_never_fills_fabric_accounting( + result_path, trajectory_path, usage, expected, trajectory_first +): from harbor.models.agent.context import AgentContext from nemo_fabric.integrations.harbor.fabric_agent import ( populate_context_from_result, @@ -352,27 +424,12 @@ def test_atif_only_fills_missing_accounting(result_path, tmp_path, usage, expect document = json.loads(result_path.read_text()) document["usage"] = usage result_path.write_text(json.dumps(document)) - trajectory = tmp_path / "trajectory.json" - trajectory.write_text( - json.dumps( - { - "schema_version": "ATIF-v1.7", - "session_id": "runtime-1", - "agent": {"name": "fabric", "version": "0.1.0"}, - "steps": [{"step_id": 1, "source": "agent", "message": "done"}], - "final_metrics": { - "total_prompt_tokens": 12, - "total_cached_tokens": 3, - "total_completion_tokens": 4, - "total_cost_usd": 0.25, - }, - } - ) - ) context = AgentContext() + if trajectory_first: + populate_context_from_trajectory(context, trajectory_path) populate_context_from_result(context, result_path) - populate_context_from_trajectory(context, trajectory) - populate_context_from_trajectory(context, trajectory) + populate_context_from_trajectory(context, trajectory_path) + populate_context_from_trajectory(context, trajectory_path) assert ( context.n_input_tokens, @@ -380,3 +437,112 @@ def test_atif_only_fills_missing_accounting(result_path, tmp_path, usage, expect context.n_output_tokens, context.cost_usd, ) == expected + assert context.metadata["fabric"]["atif_final_metrics"] == { + "source": "atif", + "coverage": "unverified", + "final_metrics": ATIF_FINAL_METRICS, + } + + +@pytest.mark.parametrize( + "passes", + [ + ("trajectory", "result"), + ("result", "trajectory", "result"), + ("trajectory", "failure"), + ("failure", "trajectory", "result"), + ("result", "trajectory", "failure"), + ], +) +def test_result_pass_keeps_atif_evidence( + result_path, failure_path, trajectory_path, passes +): + from harbor.models.agent.context import AgentContext + from nemo_fabric.integrations.harbor.fabric_agent import ( + populate_context_from_result, + populate_context_from_trajectory, + ) + + paths = {"result": result_path, "failure": failure_path} + context = AgentContext() + for name in passes: + if name == "trajectory": + populate_context_from_trajectory(context, trajectory_path) + else: + populate_context_from_result(context, paths[name]) + + fabric = context.metadata["fabric"] + assert fabric["atif_final_metrics"] == { + "source": "atif", + "coverage": "unverified", + "final_metrics": ATIF_FINAL_METRICS, + } + assert fabric["harbor_atif_validation"] == {"status": "succeeded", "error": None} + # Only the last pass's result fields remain. + if passes[-1] == "result": + assert fabric["invocation_id"] == "invocation-1" + assert "runner_error" not in fabric + else: + assert fabric["runner_error"]["code"] == "configuration_failed" + assert "invocation_id" not in fabric + + +@pytest.mark.parametrize("last_pass", ["result", "failure"]) +def test_result_pass_keeps_telemetry_validation( + result_path, failure_path, tmp_path, last_pass +): + from harbor.models.agent.context import AgentContext + from nemo_fabric.integrations.harbor.fabric_agent import ( + populate_context_from_result, + populate_context_from_telemetry_summary, + ) + + summary = {"status": "succeeded", "atof": {"records": 7}} + summary_path = tmp_path / "telemetry-validation.json" + summary_path.write_text(json.dumps(summary)) + context = AgentContext() + populate_context_from_result(context, result_path) + populate_context_from_telemetry_summary(context, summary_path) + + populate_context_from_result( + context, {"result": result_path, "failure": failure_path}[last_pass] + ) + + assert context.metadata["fabric"]["telemetry_validation"] == summary + + +@pytest.mark.parametrize("prepopulated", [False, True]) +@pytest.mark.parametrize("runner_failure_document", [False, True]) +def test_no_fabric_result_keeps_atif_out_of_accounting( + failure_path, trajectory_path, runner_failure_document, prepopulated +): + from harbor.models.agent.context import AgentContext + from nemo_fabric.integrations.harbor.fabric_agent import FabricAgent + + agent = FabricAgent(logs_dir=failure_path.parent, fabric_adapter_id="acme.agent") + if runner_failure_document: + agent._result_path = failure_path + context = ( + AgentContext( + n_input_tokens=12, n_cache_tokens=3, n_output_tokens=4, cost_usd=0.25 + ) + if prepopulated + else AgentContext() + ) + + agent.populate_context_post_run(context) + + assert ( + context.n_input_tokens, + context.n_cache_tokens, + context.n_output_tokens, + context.cost_usd, + ) == (None, None, None, None) + fabric = context.metadata["fabric"] + assert fabric["atif_final_metrics"] == { + "source": "atif", + "coverage": "unverified", + "final_metrics": ATIF_FINAL_METRICS, + } + assert fabric["harbor_atif_validation"] == {"status": "succeeded", "error": None} + assert ("runner_error" in fabric) is runner_failure_document diff --git a/tests/python/test_harbor_integration.py b/tests/python/test_harbor_integration.py index a61e43567..663006f70 100644 --- a/tests/python/test_harbor_integration.py +++ b/tests/python/test_harbor_integration.py @@ -527,7 +527,7 @@ async def test_harbor_custom_install_preserves_explicit_adapter_python(tmp_path: } -def test_harbor_populates_usage_from_canonical_atif(tmp_path: Path): +def test_harbor_keeps_canonical_atif_usage_separate(tmp_path: Path): from nemo_fabric.integrations.harbor.fabric_agent import ( populate_context_from_trajectory, ) @@ -554,10 +554,22 @@ def test_harbor_populates_usage_from_canonical_atif(tmp_path: Path): populate_context_from_trajectory(context, trajectory) - assert context.n_input_tokens == 12 - assert context.n_cache_tokens == 3 - assert context.n_output_tokens == 4 - assert context.cost_usd == 0.25 + assert context.n_input_tokens is None + assert context.n_cache_tokens is None + assert context.n_output_tokens is None + assert context.cost_usd is None + assert context.metadata["fabric"]["atif_final_metrics"] == { + "source": "atif", + "coverage": "unverified", + "final_metrics": { + "total_prompt_tokens": 12, + "total_completion_tokens": 4, + "total_cached_tokens": 3, + "total_cost_usd": 0.25, + "total_steps": None, + "extra": None, + }, + } assert context.metadata["fabric"]["harbor_atif_validation"] == { "status": "succeeded", "error": None,