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 a933ee25b..c502650a6 100644 --- a/docs/config.md +++ b/docs/config.md @@ -2391,6 +2391,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. + #### Inference origin header In external mode, Nerve also adds one `X-Nerve-Inference-Origin` header to diff --git a/nerve/agent/backends/base.py b/nerve/agent/backends/base.py index d2ddc69ed..3ee5a1a5d 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 # The tag of the session's origin for the same proxy, for example the # cron job (nerve.session_origin.origin_tag), in external mode. None: no # origin. A backend that cannot send it ignores it. diff --git a/nerve/agent/backends/claude.py b/nerve/agent/backends/claude.py index 417ace0a6..035600c99 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, inference_origin=spec.inference_origin, ), cwd=spec.cwd, @@ -843,13 +844,16 @@ 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_origin: str | None = None, + inference_trigger: str | None = None, + inference_cron_job: str | None = None, + inference_origin: str | None = None, ) -> dict[str, str]: """Build environment variables for the SDK subprocess. ``inference_principal``, ``inference_trigger`` and - ``inference_origin``: the principal, the trigger and the origin that - every model request of the process names (external mode; see + ``inference_cron_job`` and ``inference_origin``: the principal, the + trigger, the cron job and the origin that every model request of the + process names (external mode; see nerve.inference_principal). """ config = self.config @@ -926,7 +930,9 @@ def _build_env( # 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, origin=inference_origin, + inference_principal, trigger=inference_trigger, + cron_job=inference_cron_job, + origin=inference_origin, )) return env diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 111b849af..d2985686a 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -99,6 +99,7 @@ resolve_cause, ) from nerve.inference_principal import ( + inference_cron_job, inference_origin, inference_principal, inference_trigger, @@ -111,7 +112,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 @@ -1778,6 +1779,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, @@ -1805,12 +1809,14 @@ 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), ), # Where the session comes from (its cron job, channel, ...), # also fixed for the session. - inference_origin=inference_origin(session_origin(session or {})), + inference_origin=inference_origin(origin), extra=( # Per-session codex sandbox override (review-loop verifier # legs run workspace-write instead of the global default). diff --git a/nerve/inference_principal.py b/nerve/inference_principal.py index e7c089956..afa18ad87 100644 --- a/nerve/inference_principal.py +++ b/nerve/inference_principal.py @@ -1,4 +1,5 @@ -"""The principal, trigger and origin that Nerve names on model requests in external mode. +"""The principal, trigger, cron job and origin 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,7 +15,13 @@ :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-Origin``, names where a session comes +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. + +Another header, ``X-Nerve-Inference-Origin``, names where a session comes from: the cron job, the channel conversation, the hook, the plan, the workflow run or the external client, as a tag such as ``cron=`` (:func:`nerve.session_origin.origin_tag`). Like the trigger, it comes from @@ -22,24 +29,25 @@ without such a detail, and Nerve's own model calls, send no origin header. Local mode sends no header, and a NULL cause sends no principal header: Nerve -also removes a principal line, a trigger line and an origin line from the -``ANTHROPIC_CUSTOM_HEADERS`` that its own environment gives the CLI. +also removes inherited principal, trigger, cron job and origin lines from +``ANTHROPIC_CUSTOM_HEADERS`` in the environment it 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, trigger and origin never change. +enough, because the session's cause, trigger, cron job and origin never change. Codex sends the principal header only to ``codex.base_url``, and sends no -trigger or origin 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. +trigger, cron job or origin 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. """ from __future__ import annotations +import hashlib import logging import os import re @@ -58,6 +66,7 @@ INFERENCE_PRINCIPAL_HEADER = "X-Nerve-Inference-Principal" INFERENCE_TRIGGER_HEADER = "X-Nerve-Inference-Trigger" +INFERENCE_CRON_JOB_HEADER = "X-Nerve-Inference-Cron-Job" INFERENCE_ORIGIN_HEADER = "X-Nerve-Inference-Origin" CUSTOM_HEADERS_ENV = "ANTHROPIC_CUSTOM_HEADERS" @@ -65,9 +74,17 @@ # for each of them. _NERVE_HEADERS = frozenset({ INFERENCE_PRINCIPAL_HEADER.lower(), INFERENCE_TRIGGER_HEADER.lower(), + INFERENCE_CRON_JOB_HEADER.lower(), INFERENCE_ORIGIN_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") @@ -125,6 +142,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 inference_origin(origin: Origin) -> str | None: """The origin to name on the model requests of a session with ``origin``. @@ -178,13 +225,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, the trigger or the origin header gets + Each such name of the principal, trigger, cron job or origin 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. It never names - an origin. + a cron job or an origin. """ from anthropic import omit @@ -204,21 +251,23 @@ def direct_client_headers( def custom_headers_env( principal: str | None, environ: Mapping[str, str] | None = None, - *, trigger: str | None = None, origin: str | None = None, + *, trigger: str | None = None, cron_job: str | None = None, + origin: str | None = None, ) -> dict[str, str]: """Environment entries that make a Claude CLI process name ``principal``, - ``trigger`` and ``origin``. + ``trigger``, ``cron_job`` and ``origin``. ``principal`` comes from :func:`inference_principal`, ``trigger`` from - :func:`inference_trigger` and ``origin`` from :func:`inference_origin`. + :func:`inference_trigger`, ``cron_job`` from :func:`inference_cron_job` + and ``origin`` from :func:`inference_origin`. 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, the trigger or the - origin header, plus, in external mode, one line for each of - ``principal``, ``trigger`` and ``origin`` 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. + cron job or origin header, plus, in external mode, one line for each of + ``principal``, ``trigger``, ``cron_job`` and ``origin`` 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, "") @@ -226,7 +275,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 = origin = None + principal = trigger = cron_job = origin = None kept = [ line for line in lines if line.strip() and _header_name(line) not in _NERVE_HEADERS @@ -236,6 +285,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}") if is_origin_tag(origin): kept.append(f"{INFERENCE_ORIGIN_HEADER}: {origin}") value = "\n".join(kept) 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 90a2ef158..710f5081d 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,10 +35,13 @@ from nerve.identity import Actor from nerve.inference_principal import ( CUSTOM_HEADERS_ENV, + INFERENCE_CRON_JOB_HEADER, INFERENCE_ORIGIN_HEADER, INFERENCE_PRINCIPAL_HEADER, INFERENCE_TRIGGER_HEADER, + cron_job_tag_value, custom_headers_env, + inference_cron_job, direct_client_headers, inference_origin, inference_principal, @@ -87,6 +91,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] + + def _origins(env: dict[str, str]) -> list[str]: return [ value for name, value in _cli_headers(env.get(CUSTOM_HEADERS_ENV, "")) @@ -152,6 +167,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 TestOrigin: def test_local_mode_names_no_origin(self): assert inference_origin(cron_origin("nightly")) is None @@ -253,6 +326,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_one_header_line_for_the_origin(self, external): env = custom_headers_env( ALICE.actor_id, {}, trigger="cron", origin="cron=nightly", @@ -525,7 +624,8 @@ 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\nx-nerve-inference-origin: hook=other" + "x-nerve-inference-trigger: api\nx-nerve-inference-cron-job: other\n" + "x-nerve-inference-origin: hook=other" ) @@ -569,6 +669,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_each_session_names_its_origin(external, engine, cli, db): await engine.run_cron(job_id="nightly", prompt="tidy up") @@ -822,6 +954,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 _origin_values(headers: list[tuple[str, str]]) -> list[str]: return [ value for name, value in headers @@ -850,6 +989,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 _origin_values(headers) == [] assert ("x-proxy-tag", "blue") in headers @@ -862,6 +1002,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 _origin_values(headers) == [] assert ("x-proxy-tag", "blue") in headers @@ -944,6 +1085,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) == [] assert _origin_values(headers) == [] 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; }