fix(observability): attribute ACP and batch execution surfaces

Fleet telemetry showed "unknown" as the single largest execution_surface
bucket. Two construction paths were mis-attributed, both silently:

1. ACP editor sessions (VS Code / Zed / JetBrains) declare platform="acp",
   but "acp" was absent from EXECUTION_SURFACES, so the contract's
   closed-schema fallback folded every editor session into "other" --
   the bucket meant for genuinely unclassifiable traffic.

2. batch_runner built agents from _AGENT_PASSTHROUGH, which omitted
   "platform" entirely, so every batch task run reported "unknown"
   despite "batch" already being a first-class surface.

Neither is a reporting bug in the exporter: both are declaration gaps at
the construction site. "unknown" must mean "this run genuinely could not
be attributed", not "a construction site forgot to say who it was".

Changes:
- add "acp" to EXECUTION_SURFACES and map it to the "interactive"
  entrypoint alongside cli/desktop/tui
- add "acp" to the v2 wire schema enum (kept in sync by an existing test)
- pass platform through batch_runner: added to _AGENT_PASSTHROUGH, set
  self.platform = "batch" on the runner, and defaulted at the worker call
  site so callers that build a config without it stay attributable

Wire compatibility: the ingest service validates the envelope only and
stores metric bodies verbatim, so packages carrying the new value are
accepted by the already-deployed server. No coordinated deploy needed.

Tests: 12 new behavioural tests. Verified red before the fix (4 failed),
green after. Three fix-mutants confirmed killed:
  M1 revert acp from EXECUTION_SURFACES  -> 3 failed
  M2 revert acp entrypoint mapping only  -> 1 failed
  M3 revert batch passthrough            -> 1 failed
No source-text assertions; every test is a contract between the surfaces
the schema accepts and the surface each path declares. A guard test pins
that a genuinely undeclared run still reports "unknown", so attribution
cannot be "fixed" by inventing a default that hides real gaps.

batch_runner.py

@@ -58,6 +58,9 @@ _AGENT_PASSTHROUGH = (
     "base_url", "api_key", "ephemeral_system_prompt", "providers_allowed", "providers_ignored",
     "providers_order", "provider_sort", "openrouter_min_coding_score",
     "reasoning_config", "prefill_messages",
+    # Without this, every batch task run is attributed to the "unknown" execution
+    # surface in shared metrics even though "batch" is a first-class surface.
+    "platform",
 )
 
 
@@ -251,7 +254,11 @@ def _process_single_prompt(
             log_prefix=f"[B{batch_num}:P{prompt_index}]",
             skip_context_files=True,  # Don't pollute trajectories with SOUL.md/AGENTS.md
             skip_memory=True,  # Don't use persistent memory in batch runs
-            **{key: config.get(key) for key in _AGENT_PASSTHROUGH},
+            **{key: config.get(key) for key in _AGENT_PASSTHROUGH if key != "platform"},
+            # Batch is a first-class execution surface. Defaulting here (rather than
+            # relying on the caller's config dict) keeps task-run telemetry attributable
+            # even for callers that build a config without it.
+            platform=config.get("platform") or "batch",
         )
 
         # task_id ensures each task gets its own isolated VM
@@ -442,6 +449,9 @@ class BatchRunner:
         self.dataset_file = Path(dataset_file)
         for name in _RUNNER_FIELDS:
             setattr(self, name, params[name])
+        # Batch runs are their own execution surface; declaring it here keeps every
+        # worker's task-run telemetry attributable instead of falling back to "unknown".
+        self.platform = "batch"
 
         if not validate_distribution(distribution):
             raise ValueError(f"Unknown distribution: {distribution}. Available: {list(list_distributions().keys())}")

hermes_cli/observability/schemas/hermes.shared_metrics.v2.schema.json

@@ -391,6 +391,7 @@
     },
     "execution_surface": {
       "enum": [
+        "acp",
         "api",
         "batch",
         "cli",

hermes_cli/observability/shared_metrics_contract.py

@@ -38,8 +38,8 @@ _METRIC_IDENTIFIER_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789.
 _METRIC_IDENTIFIER_START_CHARACTERS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789")
 
 EXECUTION_SURFACES = frozenset({
-    "api", "batch", "cli", "desktop", "gateway", "python", "scheduled_task", "tui", "other",
+    "acp", "api", "batch", "cli", "desktop", "gateway", "python", "scheduled_task", "tui",
-    "unknown",
+    "other", "unknown",
 })
 TASK_OUTCOMES = frozenset({"cancelled", "failed", "success", "timed_out", "unknown"})
 TASK_END_REASONS = frozenset({
@@ -420,7 +420,9 @@ def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]:
 
 
 _SURFACE_ENTRYPOINTS = {
-    **dict.fromkeys(("cli", "desktop", "tui"), "interactive"),
+    # An ACP session is a human in an editor (VS Code / Zed / JetBrains), same
+    # dispatch shape as the other interactive surfaces.
+    **dict.fromkeys(("acp", "cli", "desktop", "tui"), "interactive"),
     **{s: s for s in ("api", "batch", "python", "scheduled_task", "unknown")},
     "gateway": "gateway_message",
 }

tests/hermes_cli/test_shared_metrics_surface_attribution.py

@@ -0,0 +1,77 @@
+"""Every AIAgent construction path must attribute its execution surface.
+
+The shared-metrics contract normalises a missing or unrecognised ``platform``
+to ``unknown`` / ``other``. That is the correct behaviour for a *bounded*
+schema, but it means a construction site that forgets to declare its surface
+is silently mis-attributed rather than loudly broken: fleet telemetry then
+reports "unknown" for real, attributable traffic.
+
+These are contract tests between two pieces of data -- the surfaces the
+contract accepts, and the surface each construction path actually declares --
+not snapshots of any current value.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from hermes_cli.observability import shared_metrics_contract as contract
+
+
+def test_acp_editor_sessions_get_their_own_surface():
+    """ACP (VS Code / Zed / JetBrains) is a real interactive surface, not 'other'.
+
+    The ACP adapter declares ``platform="acp"``. If that value is not an
+    accepted surface, the contract's closed-schema fallback buckets every
+    editor session into ``other`` alongside genuinely unclassifiable traffic.
+    """
+    assert contract.execution_surface({"platform": "acp"}) == "acp"
+
+
+def test_acp_sessions_are_interactive():
+    """An editor session is a human at a keyboard, like cli/tui/desktop."""
+    fields = contract.task_start_fields({"platform": "acp"})
+    assert fields["entrypoint"] == "interactive"
+    assert fields["execution_surface"] == "acp"
+
+
+def test_batch_runs_declare_their_surface():
+    """batch_runner builds agents from a fixed passthrough tuple.
+
+    ``batch`` is already an accepted surface, so the only defect is that the
+    runner never declares it -- every batch task run reports 'unknown'.
+    """
+    import batch_runner
+
+    assert "platform" in batch_runner._AGENT_PASSTHROUGH, (
+        "batch_runner._AGENT_PASSTHROUGH omits 'platform', so batch task runs are "
+        f"attributed to {contract.execution_surface({})!r} despite 'batch' being a "
+        "valid execution surface"
+    )
+
+
+@pytest.mark.parametrize(
+    "platform",
+    ["cli", "tui", "desktop", "batch", "acp", "api_server", "cron", "telegram"],
+)
+def test_declared_platforms_resolve_to_a_named_surface(platform):
+    """No production construction path should resolve to unknown/other.
+
+    'unknown' must mean "this run genuinely could not be attributed", not
+    "a construction site forgot to say who it was".
+    """
+    surface = contract.execution_surface({"platform": platform})
+    assert surface not in {"unknown", "other"}, (
+        f"platform={platform!r} resolves to {surface!r}; a real surface is being "
+        "folded into the catch-all bucket"
+    )
+
+
+def test_unattributed_runs_still_report_unknown():
+    """The catch-all must survive: a genuinely undeclared run is 'unknown'.
+
+    This is the counterpart to the tests above -- fixing attribution must not
+    be achieved by inventing a default that hides real gaps.
+    """
+    assert contract.execution_surface({}) == "unknown"
+    assert contract.task_start_fields({})["entrypoint"] == "unknown"