From ab85e8aab71fffa139593496b6494a097c5b4afe Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:29:09 +0000 Subject: [PATCH] Name the cron job on the model requests of its sessions In external mode, a proxy in front of the model API can attribute model usage by person and by trigger, but not by cron job. Scheduled work is a large part of an agent's model usage, so the cost of each job must be visible. The model requests of a session that a cron job started now carry one X-Nerve-Inference-Cron-Job header with the job, through ANTHROPIC_CUSTOM_HEADERS, under the same rules as the principal and trigger headers. Job IDs are free text, so Nerve changes the ID to 1 to 64 bytes of ASCII letters, digits, '.', '_' and '-', with a short hash when the ID has no such character or is too long. The leg session of a workflow run that a cron job started records the job in its origin, so its requests name the job too. The origin link of such a session opens the workflow runs, as for other workflow legs. Nerve's own model calls, Run later sessions and Codex send no cron job header. Co-Authored-By: Claude Opus 5.5 --- docs/accounts.md | 5 +- docs/config.md | 27 ++++++ nerve/agent/backends/base.py | 4 + nerve/agent/backends/claude.py | 14 ++- nerve/agent/engine.py | 17 +++- nerve/inference_principal.py | 84 ++++++++++++---- nerve/workflows/service.py | 25 ++++- tests/test_inference_principal.py | 143 +++++++++++++++++++++++++++- tests/test_workflow_attribution.py | 14 +++ web/src/utils/sessionOrigin.test.ts | 6 ++ web/src/utils/sessionOrigin.ts | 2 +- 11 files changed, 311 insertions(+), 30 deletions(-) diff --git a/docs/accounts.md b/docs/accounts.md index 231b03df5..a5cb0f41f 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -197,7 +197,10 @@ In external mode: - Each model request names the principal that caused the work, in the `X-Nerve-Inference-Principal` header: the session's cause, or the system actor for Nerve's own model calls. See - [Inference principal header](config.md#inference-principal-header). + [Inference principal header](config.md#inference-principal-header). The + model requests of a cron job's session also name the job, in the + `X-Nerve-Inference-Cron-Job` header. See + [Inference cron job header](config.md#inference-cron-job-header). > **Warning:** Without `auth.external.gateway_jwks_file`, Nerve trusts the > header. Then use external mode only when the gateway is the only caller that diff --git a/docs/config.md b/docs/config.md index 336f37682..e208d5d37 100644 --- a/docs/config.md +++ b/docs/config.md @@ -2390,6 +2390,33 @@ and a CLI process keeps it for its whole life. Nerve does not prove the value. The same requests as above have no header, and Codex requests to `codex.base_url` have no trigger header. +#### Inference cron job header + +In external mode, Nerve also adds one `X-Nerve-Inference-Cron-Job` header to +each model request of a session that a cron job started: the sessions of +isolated and persistent cron jobs, and the session of a workflow run that a +cron job started. The value names the job, so that a proxy in front of the +model API can attribute model usage to it. Other sessions ("Run later" too) +and Nerve's own model calls send no cron job header. + +Cron job IDs are free text, and the value must be a short, safe header +value. So Nerve changes the job ID: + +1. Each run of characters other than ASCII letters, digits, `.`, `_` and `-` + becomes one `-`. +2. Nerve removes `-` at the start and at the end. +3. An empty result becomes `job-` and the first 8 hex digits of the SHA-256 + of the job ID (UTF-8). +4. A result longer than 64 bytes becomes its first 55 bytes, `-`, and the + same 8 hex digits. + +For example, `daily-digest` stays `daily-digest`, and `Morning briefing` +becomes `Morning-briefing`. The value is 1 to 64 bytes. The header has the +same rules as the trigger header: local mode sends none, Nerve removes an +inherited line for it from `ANTHROPIC_CUSTOM_HEADERS`, and a CLI process keeps +it for its whole life. Nerve does not prove the value. Codex requests have no +cron job header. + ### System principal ID The environment variable `NERVE_SYSTEM_PRINCIPAL_ID` gives the ID of the diff --git a/nerve/agent/backends/base.py b/nerve/agent/backends/base.py index 53cd3ff45..835cf7f50 100644 --- a/nerve/agent/backends/base.py +++ b/nerve/agent/backends/base.py @@ -90,6 +90,10 @@ class SessionSpec: # The session's trigger for the same proxy (nerve.session_origin), in # external mode. None: no trigger. A backend that cannot send it ignores it. inference_trigger: str | None = None + # The header value of the session's cron job for the same proxy + # (nerve.inference_principal.inference_cron_job), in external mode. None: + # no cron job. A backend that cannot send it ignores it. + inference_cron_job: str | None = None extra: dict[str, Any] = field(default_factory=dict) diff --git a/nerve/agent/backends/claude.py b/nerve/agent/backends/claude.py index 24d931ab9..e62040cd1 100644 --- a/nerve/agent/backends/claude.py +++ b/nerve/agent/backends/claude.py @@ -738,6 +738,7 @@ def _cli_stderr(line: str) -> None: cache_ttl=spec.cache_ttl, inference_principal=spec.inference_principal, inference_trigger=spec.inference_trigger, + inference_cron_job=spec.inference_cron_job, ), cwd=spec.cwd, # Load no filesystem settings/memory: Nerve owns the entire @@ -843,12 +844,14 @@ def _sweep_system_prompt_dir(dir_path: Path) -> None: def _build_env( self, cache_ttl: str = "5m", inference_principal: str | None = None, inference_trigger: str | None = None, + inference_cron_job: str | None = None, ) -> dict[str, str]: """Build environment variables for the SDK subprocess. - ``inference_principal`` and ``inference_trigger``: the principal and - the trigger that every model request of the process names (external - mode; see nerve.inference_principal). + ``inference_principal``, ``inference_trigger`` and + ``inference_cron_job``: the principal, the trigger and the cron job + that every model request of the process names (external mode; see + nerve.inference_principal). """ config = self.config env: dict[str, str] = {} @@ -921,7 +924,10 @@ def _build_env( env[env_name] = str(target) # The SDK merges this dict over Nerve's own environment, so an # inherited ANTHROPIC_CUSTOM_HEADERS is kept and extended here. - env.update(custom_headers_env(inference_principal, trigger=inference_trigger)) + env.update(custom_headers_env( + inference_principal, trigger=inference_trigger, + cron_job=inference_cron_job, + )) return env def _build_mcp_servers(self, session_id: str) -> dict[str, Any]: diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index c8b0062f3..a384fcf66 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -98,7 +98,11 @@ attribution_for, resolve_cause, ) -from nerve.inference_principal import inference_principal, inference_trigger +from nerve.inference_principal import ( + inference_cron_job, + inference_principal, + inference_trigger, +) from nerve.mcp_gateway import McpGatewayCatalog from nerve.memory.backend import ( MemoryBackendUnavailable, @@ -107,7 +111,7 @@ ) from nerve.observability.langfuse import attributes as lf_attrs from nerve.platform.activity import LEASE_MEMORIZE, keep_awake -from nerve.session_origin import cron_origin, session_origin +from nerve.session_origin import DETAIL_CRON_JOB, cron_origin, session_origin from nerve.skills.manager import SkillManager from nerve.utils.llm import response_text @@ -1774,6 +1778,9 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: # or a newer one (see _mcp_catalog_rebuild_due). self._client_mcp_revisions[session_id] = self._mcp_catalog_revision + # The session's origin, fixed for the session: its trigger and, + # for a session of a cron job, the job. + origin = session_origin(session or {}) spec = SessionSpec( session_id=session_id, source=source, @@ -1801,8 +1808,10 @@ async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: (session or {}).get("caused_by_actor_id"), ), # The session's trigger, also fixed for the session. - inference_trigger=inference_trigger( - session_origin(session or {}).trigger, + inference_trigger=inference_trigger(origin.trigger), + # The session's cron job, if a cron job started it. + inference_cron_job=inference_cron_job( + origin.detail.get(DETAIL_CRON_JOB), ), extra=( # Per-session codex sandbox override (review-loop verifier diff --git a/nerve/inference_principal.py b/nerve/inference_principal.py index 41a407062..850cc7c30 100644 --- a/nerve/inference_principal.py +++ b/nerve/inference_principal.py @@ -1,4 +1,4 @@ -"""The principal and the trigger that Nerve names on each model request in external mode. +"""The principal, the trigger and the cron job that Nerve names on model requests in external mode. In external mode, a proxy in front of the model API can attribute model usage to a person. Nerve gives it the principal that caused the work: one @@ -14,18 +14,24 @@ :mod:`nerve.session_origin`, or ``system`` for Nerve's own model calls. Like the cause, the trigger of a session is set when the session is created. +A third header, ``X-Nerve-Inference-Cron-Job``, names the cron job of a +session that a cron job started (``origin_detail.cron_job``), so that the +proxy can attribute model usage to the job. Job IDs are free text, so Nerve +sends :func:`cron_job_tag_value` of the ID: 1 to 64 bytes of ASCII letters, +digits, ``.``, ``_`` and ``-``. Nerve's own model calls send no cron job. + Local mode sends no header, and a NULL cause sends no principal header: Nerve -also removes a principal line and a trigger line from the +also removes a principal line, a trigger line and a cron job line from the ``ANTHROPIC_CUSTOM_HEADERS`` that its own environment gives the CLI. The Claude CLI reads ``ANTHROPIC_CUSTOM_HEADERS`` from its environment when it builds the API client for a request: one ``Name: Value`` header per line, lines split on ``\\n`` or ``\\r\\n``, name and value split on the first ``:`` and trimmed. The environment of a CLI process is fixed when it starts. That is -enough, because the session's cause and trigger never change. +enough, because the session's cause, trigger and cron job never change. Codex sends the principal header only to ``codex.base_url``, and sends no -trigger header: the provider that Nerve gives Codex for it reads the value +trigger header and no cron job header: the provider that Nerve gives Codex for it reads the value from an environment variable of the Codex process (``env_http_headers``). See ``nerve.agent.backends.codex.backend``. One Codex process serves one session, so the same rule applies. @@ -33,6 +39,7 @@ from __future__ import annotations +import hashlib import logging import os import re @@ -45,14 +52,23 @@ INFERENCE_PRINCIPAL_HEADER = "X-Nerve-Inference-Principal" INFERENCE_TRIGGER_HEADER = "X-Nerve-Inference-Trigger" +INFERENCE_CRON_JOB_HEADER = "X-Nerve-Inference-Cron-Job" CUSTOM_HEADERS_ENV = "ANTHROPIC_CUSTOM_HEADERS" # The headers that Nerve sets, in lower case. Nerve removes an inherited line # for each of them. _NERVE_HEADERS = frozenset({ INFERENCE_PRINCIPAL_HEADER.lower(), INFERENCE_TRIGGER_HEADER.lower(), + INFERENCE_CRON_JOB_HEADER.lower(), }) +# The value rule of the cron job header, which the proxy checks again. +MAX_CRON_JOB_BYTES = 64 +_CRON_JOB_VALUE = re.compile(rf"[A-Za-z0-9._-]{{1,{MAX_CRON_JOB_BYTES}}}") +_CRON_JOB_OTHER = re.compile(r"[^A-Za-z0-9._-]+") +# Length of the hash part of a changed cron job value, in hex digits. +_CRON_JOB_HASH_HEX = 8 + # The CLI splits the variable on these line ends (and on nothing else). _LINE_END = re.compile(r"\r?\n") @@ -110,6 +126,36 @@ def inference_trigger(trigger: str | None) -> str | None: return trigger +def cron_job_tag_value(job_id: str) -> str: + """The value of the cron job header for cron job ``job_id``. + + Each run of characters other than ASCII letters, digits, ``.``, ``_`` + and ``-`` becomes one ``-``, and ``-`` at the start and the end is + removed. An empty result becomes ``job-`` and the first 8 hex digits of + the SHA-256 of ``job_id`` (UTF-8). A result longer than 64 bytes becomes + its first 55 bytes, ``-`` and the same 8 hex digits. So the value is 1 + to 64 bytes, and two different IDs give the same value only rarely. + """ + value = _CRON_JOB_OTHER.sub("-", job_id).strip("-") + if value and len(value) <= MAX_CRON_JOB_BYTES: + return value + digest = hashlib.sha256(job_id.encode("utf-8")).hexdigest()[:_CRON_JOB_HASH_HEX] + if not value: + return f"job-{digest}" + return f"{value[:MAX_CRON_JOB_BYTES - _CRON_JOB_HASH_HEX - 1]}-{digest}" + + +def inference_cron_job(job_id: str | None) -> str | None: + """The cron job to name on the model requests of a session of ``job_id``. + + ``None`` in local mode and when the session has no cron job; else + :func:`cron_job_tag_value` of the ID. + """ + if not job_id or not _is_external_mode(): + return None + return cron_job_tag_value(job_id) + + def _system_trigger_headers() -> dict[str, str]: trigger = inference_trigger(TRIGGER_SYSTEM) return {INFERENCE_TRIGGER_HEADER: trigger} if trigger else {} @@ -152,12 +198,13 @@ def direct_client_headers( The Anthropic SDK also adds the lines of ``ANTHROPIC_CUSTOM_HEADERS`` in the environment of Nerve to its default headers, with the name as written. - Each such name of the principal or the trigger header gets + Each such name of the principal, the trigger or the cron job header gets ``anthropic.omit``, which removes it. Then, in external mode, the header for ``principal_id`` is added, or for the system actor of the global database when it is ``None``, and the trigger ``system`` (:func:`system_inference_headers`). So a request names one principal and - one trigger in external mode, and neither in local mode. + one trigger in external mode, and neither in local mode. It never names + a cron job. """ from anthropic import omit @@ -177,20 +224,21 @@ def direct_client_headers( def custom_headers_env( principal: str | None, environ: Mapping[str, str] | None = None, - *, trigger: str | None = None, + *, trigger: str | None = None, cron_job: str | None = None, ) -> dict[str, str]: - """Environment entries that make a Claude CLI process name ``principal`` - and ``trigger``. + """Environment entries that make a Claude CLI process name ``principal``, + ``trigger`` and ``cron_job``. - ``principal`` comes from :func:`inference_principal` and ``trigger`` from - :func:`inference_trigger`. The SDK merges the entries over the + ``principal`` comes from :func:`inference_principal`, ``trigger`` from + :func:`inference_trigger` and ``cron_job`` from + :func:`inference_cron_job`. The SDK merges the entries over the environment of Nerve, which the CLI inherits. The value is the inherited ``ANTHROPIC_CUSTOM_HEADERS`` without blank lines and without a line for - the principal or the trigger header, plus, in external mode, one line for - each of ``principal`` and ``trigger`` that is set. So the CLI sends at - most one header of each in external mode, and none in local mode. No - entry when that value is the inherited value, and in local mode when the - inherited value has no line to remove. + the principal, the trigger or the cron job header, plus, in external + mode, one line for each of ``principal``, ``trigger`` and ``cron_job`` + that is set. So the CLI sends at most one header of each in external + mode, and none in local mode. No entry when that value is the inherited + value, and in local mode when the inherited value has no line to remove. """ env = os.environ if environ is None else environ inherited = env.get(CUSTOM_HEADERS_ENV, "") @@ -198,7 +246,7 @@ def custom_headers_env( if not _is_external_mode(): if not any(_header_name(line) in _NERVE_HEADERS for line in lines): return {} - principal = trigger = None + principal = trigger = cron_job = None kept = [ line for line in lines if line.strip() and _header_name(line) not in _NERVE_HEADERS @@ -208,6 +256,8 @@ def custom_headers_env( kept.append(f"{INFERENCE_PRINCIPAL_HEADER}: {principal}") if trigger in TRIGGERS: kept.append(f"{INFERENCE_TRIGGER_HEADER}: {trigger}") + if cron_job and _CRON_JOB_VALUE.fullmatch(cron_job): + kept.append(f"{INFERENCE_CRON_JOB_HEADER}: {cron_job}") value = "\n".join(kept) if value == inherited: return {} diff --git a/nerve/workflows/service.py b/nerve/workflows/service.py index be2bd9e07..ba07bed7e 100644 --- a/nerve/workflows/service.py +++ b/nerve/workflows/service.py @@ -59,7 +59,12 @@ from nerve.db.usage import estimate_turn_cost from nerve.db.workflow_runs import ACTIVE_STATUSES from nerve.identity import UNATTRIBUTED, Attribution, attribution_for -from nerve.session_origin import DETAIL_WORKFLOW_RUN, TRIGGER_WORKFLOW, Origin +from nerve.session_origin import ( + DETAIL_CRON_JOB, + DETAIL_WORKFLOW_RUN, + TRIGGER_WORKFLOW, + Origin, +) from nerve.utils.time import utc_now_iso if TYPE_CHECKING: @@ -85,6 +90,8 @@ _ERROR_MAX = 2000 _SESSION_PREFIX = "workflow:" +# ``created_by`` prefix of a run that a cron job started (nerve.cron.service). +_CRON_CREATED_BY = "cron:" class WorkflowRunError(ValueError): @@ -446,6 +453,20 @@ async def _origin_session_id(self, run: dict) -> str | None: return None return sid if await self.db.get_session(sid) else None + @staticmethod + def _leg_origin(run: dict) -> Origin: + """The origin of the leg session of ``run``. + + The run's ID, and for a run that a cron job started + (``created_by`` ``cron:``) the job, so that the model requests + of the leg name the job (nerve.inference_principal). + """ + detail = {DETAIL_WORKFLOW_RUN: run["id"]} + cb = str(run.get("created_by") or "") + if cb.startswith(_CRON_CREATED_BY): + detail[DETAIL_CRON_JOB] = cb[len(_CRON_CREATED_BY):] + return Origin(TRIGGER_WORKFLOW, detail) + def _default_model(self, backend: str) -> str: if backend == "codex": return self.config.codex.model @@ -540,7 +561,7 @@ async def _execute(self, run_id: str) -> None: cwd=spec.get("cwd") or None, actor=leg_actor, cause=run.get("caused_by_actor_id"), - origin=Origin(TRIGGER_WORKFLOW, {DETAIL_WORKFLOW_RUN: run_id}), + origin=self._leg_origin(run), ) if origin: await self.db.update_session_fields( diff --git a/tests/test_inference_principal.py b/tests/test_inference_principal.py index 43bc30b38..1556982bd 100644 --- a/tests/test_inference_principal.py +++ b/tests/test_inference_principal.py @@ -16,6 +16,7 @@ import asyncio import contextlib import dataclasses +import hashlib import json import logging import re @@ -34,9 +35,12 @@ from nerve.identity import Actor from nerve.inference_principal import ( CUSTOM_HEADERS_ENV, + INFERENCE_CRON_JOB_HEADER, INFERENCE_PRINCIPAL_HEADER, INFERENCE_TRIGGER_HEADER, + cron_job_tag_value, custom_headers_env, + inference_cron_job, inference_principal, inference_trigger, principal_headers, @@ -84,6 +88,17 @@ def _triggers(env: dict[str, str]) -> list[str]: ] +def _cron_jobs(env: dict[str, str]) -> list[str]: + return [ + value for name, value in _cli_headers(env.get(CUSTOM_HEADERS_ENV, "")) + if name.lower() == INFERENCE_CRON_JOB_HEADER.lower() + ] + + +def _hash8(job_id: str) -> str: + return hashlib.sha256(job_id.encode("utf-8")).hexdigest()[:8] + + @pytest.fixture def external(monkeypatch): monkeypatch.setenv(AUTH_MODE_ENV, AUTH_MODE_EXTERNAL) @@ -142,6 +157,64 @@ def test_a_value_outside_the_vocabulary_names_no_trigger(self, external, bad): assert inference_trigger(bad) is None +_CRON_JOB_RULE = re.compile(r"[A-Za-z0-9._-]{1,64}") + + +class TestCronJob: + @pytest.mark.parametrize("job_id", [ + "nightly", "daily-digest", "inbox_check", "v1.2", "A" * 64, + ]) + def test_an_id_that_obeys_the_rule_is_unchanged(self, job_id): + assert cron_job_tag_value(job_id) == job_id + + @pytest.mark.parametrize(("job_id", "value"), [ + ("Morning briefing", "Morning-briefing"), + ("check / inbox", "check-inbox"), + ("pr-review:main", "pr-review-main"), + ("caf\u00e9 report", "caf-report"), + ("a,b\nc", "a-b-c"), + (" -weekly- ", "weekly"), + ("--x--", "x"), + ]) + def test_other_characters_become_one_hyphen(self, job_id, value): + assert cron_job_tag_value(job_id) == value + + @pytest.mark.parametrize("job_id", ["\u65e5\u62a5", "---", " ", ":/,"]) + def test_an_empty_result_becomes_a_hash(self, job_id): + assert cron_job_tag_value(job_id) == f"job-{_hash8(job_id)}" + + def test_a_long_result_is_cut_and_gets_a_hash(self): + job_id = "b" * 65 + value = cron_job_tag_value(job_id) + assert value == "b" * 55 + "-" + _hash8(job_id) + assert len(value) == 64 + + def test_a_long_id_is_cut_after_the_transform(self): + job_id = "c " * 40 + value = cron_job_tag_value(job_id) + assert value == ("c-" * 28)[:55] + "-" + _hash8(job_id) + + def test_two_long_ids_with_the_same_start_differ(self): + assert cron_job_tag_value("d" * 70) != cron_job_tag_value("d" * 71) + + @pytest.mark.parametrize("job_id", [ + "nightly", "Morning briefing", "\u65e5\u62a5", "x" * 200, "a\r\nX-Injected: 1", + ]) + def test_every_value_obeys_the_rule(self, job_id): + assert _CRON_JOB_RULE.fullmatch(cron_job_tag_value(job_id)) + + def test_local_mode_names_no_cron_job(self): + assert inference_cron_job("nightly") is None + + def test_external_mode_names_the_transformed_id(self, external): + assert inference_cron_job("nightly") == "nightly" + assert inference_cron_job("Morning briefing") == "Morning-briefing" + + @pytest.mark.parametrize("job_id", [None, ""]) + def test_no_job_names_no_cron_job(self, external, job_id): + assert inference_cron_job(job_id) is None + + class TestCliEnvironment: def test_local_mode_removes_an_inherited_principal_and_keeps_the_rest(self): environ = {CUSTOM_HEADERS_ENV: ( @@ -224,6 +297,32 @@ def test_local_mode_removes_an_inherited_trigger(self): def test_a_trigger_outside_the_vocabulary_is_never_written(self, external): assert custom_headers_env(None, {}, trigger="cron\nX-Injected: 1") == {} + def test_one_header_line_for_the_cron_job(self, external): + env = custom_headers_env(None, {}, trigger="cron", cron_job="nightly") + assert _cli_headers(env[CUSTOM_HEADERS_ENV]) == [ + (INFERENCE_TRIGGER_HEADER, "cron"), (INFERENCE_CRON_JOB_HEADER, "nightly"), + ] + + def test_an_inherited_cron_job_is_replaced_or_removed(self, external): + environ = {CUSTOM_HEADERS_ENV: "X-Proxy-Tag: blue\nx-nerve-inference-cron-job: other"} + env = custom_headers_env(None, environ, cron_job="nightly") + assert _cli_headers(env[CUSTOM_HEADERS_ENV]) == [ + ("X-Proxy-Tag", "blue"), (INFERENCE_CRON_JOB_HEADER, "nightly"), + ] + assert custom_headers_env(None, environ) == {CUSTOM_HEADERS_ENV: "X-Proxy-Tag: blue"} + + def test_local_mode_removes_an_inherited_cron_job(self): + environ = {CUSTOM_HEADERS_ENV: "X-Proxy-Tag: blue\nX-Nerve-Inference-Cron-Job: other"} + assert custom_headers_env(None, environ, cron_job="nightly") == { + CUSTOM_HEADERS_ENV: "X-Proxy-Tag: blue", + } + + @pytest.mark.parametrize("bad", [ + "nightly\nX-Injected: 1", "two words", "a,b", "", "e" * 65, "caf\u00e9", + ]) + def test_a_cron_job_outside_the_rule_is_never_written(self, external, bad): + assert custom_headers_env(None, {}, cron_job=bad) == {} + def test_the_process_environment_is_the_default(self, external, monkeypatch): monkeypatch.setenv(CUSTOM_HEADERS_ENV, "X-Proxy-Tag: blue") env = custom_headers_env(ALICE.actor_id) @@ -461,7 +560,7 @@ async def _open_process(cmd, **kwargs): _INHERITED = ( f"X-Proxy-Tag: blue\nx-nerve-inference-principal: {BOB.actor_id}\n" - "x-nerve-inference-trigger: api" + "x-nerve-inference-trigger: api\nx-nerve-inference-cron-job: other" ) @@ -505,6 +604,38 @@ async def test_each_session_names_its_trigger(external, engine, cli, db): assert cli[1].spec.inference_trigger == "slack" +@pytest.mark.asyncio +async def test_each_session_of_a_cron_job_names_its_job(external, engine, cli, db): + await engine.run_cron(job_id="nightly", prompt="tidy up") + await engine.run_persistent_cron(job_id="Morning briefing", prompt="brief me") + await _turn(engine, "s1", ALICE) + # The leg of a workflow run that a cron job started. + await db.create_session( + "workflow:r1", source="workflow", actor=ALICE, + origin=Origin("workflow", {"workflow_run": "r1", "cron_job": "weekly"}), + ) + await _turn(engine, "workflow:r1", ALICE) + + assert [_cron_jobs(process.env) for process in cli] == [ + ["nightly"], ["Morning-briefing"], [], ["weekly"], + ] + assert [process.spec.inference_cron_job for process in cli] == [ + "nightly", "Morning-briefing", None, "weekly", + ] + assert _triggers(cli[3].env) == ["workflow"] + + +@pytest.mark.asyncio +async def test_a_cron_session_names_no_job_in_local_mode(engine, cli, monkeypatch): + monkeypatch.setenv(CUSTOM_HEADERS_ENV, _INHERITED) + await engine.run_cron(job_id="nightly", prompt="tidy up") + + (process,) = cli + assert process.spec.inference_cron_job is None + env = await _process_env(process.options, monkeypatch) + assert _cli_headers(env[CUSTOM_HEADERS_ENV]) == [("X-Proxy-Tag", "blue")] + + @pytest.mark.asyncio async def test_local_mode_sends_no_header(engine, cli): await _turn(engine, "s1", ALICE) @@ -729,6 +860,13 @@ def _trigger_values(headers: list[tuple[str, str]]) -> list[str]: ] +def _cron_job_values(headers: list[tuple[str, str]]) -> list[str]: + return [ + value for name, value in headers + if name.lower() == INFERENCE_CRON_JOB_HEADER.lower() + ] + + def test_direct_requests_in_local_mode_drop_an_inherited_principal( global_db, monkeypatch, ): @@ -737,6 +875,7 @@ def test_direct_requests_in_local_mode_drop_an_inherited_principal( headers = _sent_headers(client) assert _principal_values(headers) == [] assert _trigger_values(headers) == [] + assert _cron_job_values(headers) == [] assert ("x-proxy-tag", "blue") in headers @@ -748,6 +887,7 @@ def test_direct_requests_in_external_mode_name_the_system_actor_once( headers = _sent_headers(client) assert _principal_values(headers) == [global_db.system_actor_id] assert _trigger_values(headers) == ["system"] + assert _cron_job_values(headers) == [] assert ("x-proxy-tag", "blue") in headers @@ -829,6 +969,7 @@ async def test_the_doctor_probe_names_the_stored_system_actor( (headers,) = _MessagesApi.received assert _principal_values(headers) == [system_actor_id] assert _trigger_values(headers) == ["system"] + assert _cron_job_values(headers) == [] @pytest.mark.asyncio diff --git a/tests/test_workflow_attribution.py b/tests/test_workflow_attribution.py index 88253fd77..51c719aa5 100644 --- a/tests/test_workflow_attribution.py +++ b/tests/test_workflow_attribution.py @@ -470,6 +470,20 @@ async def test_a_run_without_a_cause_gets_a_session_without_one(install): assert await install.created_event(session_id) == (install.system_actor_id, None) +@pytest.mark.asyncio +async def test_the_session_of_a_run_that_a_cron_job_started_names_the_job(install): + run = await install.runs.start_run( + ENGINE_CLAUDE, {"prompt": "p"}, 1.0, created_by="cron:weekly-report", + ) + await install.drain() + + session = await install.db.get_session(f"workflow:{run['id']}") + assert session["trigger"] == "workflow" + assert json.loads(session["origin_detail"]) == { + "workflow_run": run["id"], "cron_job": "weekly-report", + } + + @pytest.mark.asyncio async def test_a_rest_kill_records_the_person(install): alice, bob = install.actor_id("alice"), install.actor_id("bob") diff --git a/web/src/utils/sessionOrigin.test.ts b/web/src/utils/sessionOrigin.test.ts index 5e90c4422..fc04656b9 100644 --- a/web/src/utils/sessionOrigin.test.ts +++ b/web/src/utils/sessionOrigin.test.ts @@ -77,6 +77,12 @@ describe('originLink', () => { .toEqual({ to: '/workflow-runs', label: 'Open workflow runs' }); }); + it('links the workflow runs for a run that a cron job started', () => { + const session = { trigger: 'workflow' as const, origin_detail: { workflow_run: 'wfr-1', cron_job: 'weekly' } }; + expect(originLink(session)).toEqual({ to: '/workflow-runs', label: 'Open workflow runs' }); + expect(originPhrase(session)).toBe('From cron job weekly'); + }); + it('has no link for a channel', () => { expect(originLink({ origin_detail: { channel: 'C1', channel_name: 'general' } })).toBeNull(); expect(originLink({})).toBeNull(); diff --git a/web/src/utils/sessionOrigin.ts b/web/src/utils/sessionOrigin.ts index 442bbbcd9..ce1134c23 100644 --- a/web/src/utils/sessionOrigin.ts +++ b/web/src/utils/sessionOrigin.ts @@ -73,7 +73,7 @@ export function originPhrase(session: OriginFields): string { export function originLink(session: OriginFields): { to: string; label: string } | null { const detail = session.origin_detail ?? {}; if (detail.plan_id) return { to: `/plans/${encodeURIComponent(detail.plan_id)}`, label: 'Open plan' }; - if (detail.cron_job) return { to: '/cron', label: 'Open cron jobs' }; if (detail.workflow_run) return { to: '/workflow-runs', label: 'Open workflow runs' }; + if (detail.cron_job) return { to: '/cron', label: 'Open cron jobs' }; return null; }