From 731961a3a19f33f73adad4f7417c57bf4b0497a3 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:54:20 +0000 Subject: [PATCH 1/9] Record who schedules wakeups and starts cron runs, and for whom A scheduled turn must run for the principal that caused it, so that its model requests and tool changes name that person. Until now a run-later message, an agent wakeup and a "Run now" cron run all ran for the system actor. Migration v057 adds session_wakeups.created_by_actor_id and .caused_by_actor_id, and cron_logs.actor_id and .caused_by_actor_id (nullable foreign keys to actor_refs, no backfill). The registry lists the four columns. - Run later through REST records the person as scheduler and cause. - An agent wakeup (the ScheduleWakeup hook of the Claude CLI, or the schedule_wakeup tool) records the system actor and the session's cause. - A fired wakeup runs in its session, so its turn has the session's cause. - Scheduled, catch-up and missed cron runs log the system actor as actor and cause. POST /api/cron/jobs/{id}/trigger logs the person, and a session that the run creates (an isolated run, or a new persistent generation) has that person as its cause. - Session list rows and the session_running event carry pending_wakeup_created_by_actor_id and pending_wakeup_caused_by_actor_id. Co-Authored-By: Claude Opus 5.5 --- nerve/agent/engine.py | 57 +- nerve/agent/sessions.py | 5 +- nerve/agent/tools/handlers/wakeups.py | 5 +- nerve/cron/service.py | 88 ++- nerve/db/actor_columns.py | 4 + nerve/db/cron.py | 26 +- .../migrations/v057_schedule_attribution.py | 62 ++ nerve/db/wakeups.py | 22 +- nerve/gateway/routes/cron.py | 15 +- nerve/gateway/routes/sessions.py | 18 +- tests/test_cron.py | 16 +- tests/test_schedule_attribution.py | 703 ++++++++++++++++++ tests/test_workflow_cron.py | 3 +- 13 files changed, 966 insertions(+), 58 deletions(-) create mode 100644 nerve/db/migrations/v057_schedule_attribution.py create mode 100644 tests/test_schedule_attribution.py diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 54238fc84..d34913230 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -1064,9 +1064,14 @@ def _wakeup_fire_at(cls, delay_seconds: Any) -> str: @classmethod async def _record_wakeup( cls, db: Any, session_id: str, tool_input: dict, + *, attribution: Attribution = UNATTRIBUTED, ) -> int | None: """Persist a ScheduleWakeup request from its tool input. + ``attribution`` is that of the tool call: the system actor, and the + cause of the turn that made the call. The wakeup turn runs for that + cause. + Returns the new wakeup id, or ``None`` when there's no prompt to re-inject (in which case nothing is scheduled). """ @@ -1077,6 +1082,7 @@ async def _record_wakeup( fire_at = cls._wakeup_fire_at(tool_input.get("delaySeconds")) wakeup_id = await db.add_wakeup( session_id, prompt=prompt, fire_at=fire_at, reason=reason, + attribution=attribution, ) logger.info( "Recorded wakeup %s for session %s at %s", @@ -1352,7 +1358,14 @@ async def _get_or_create_client( ) async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: - return await self._record_wakeup(self.db, sid, tool_input) + # The CLI runs the hook inside the turn that called the tool, + # so the session's turn cause is the cause of that call. + return await self._record_wakeup( + self.db, sid, tool_input, + attribution=attribution_for( + self.db.system_actor, self.session_cause(sid), + ), + ) spec = SessionSpec( session_id=session_id, @@ -1566,13 +1579,13 @@ def has_live_background_tasks(self, session_id: str) -> bool: runs to defer the terminal 'done' until spend-relevant work ends.""" return self._has_live_background_tasks(session_id) - async def pending_wakeup_at(self, session_id: str) -> str | None: - """``fire_at`` of the session's pending wakeup, or ``None``. + async def pending_wakeup(self, session_id: str) -> dict | None: + """The session's pending wakeup row, or ``None``. ``ScheduleWakeup`` keeps at most one pending row per session (a new - request supersedes the old one), so this is a single timestamp - rather than a list. Never raises — a listing/indicator caller must - not fail on an enrichment read. + request supersedes the old one), so this is a single row rather + than a list. Never raises — a listing/indicator caller must not fail + on an enrichment read. """ try: rows = await self.db.list_pending_wakeups(session_id) @@ -1581,7 +1594,12 @@ async def pending_wakeup_at(self, session_id: str) -> str | None: "pending wakeup lookup failed for %s", session_id, exc_info=True, ) return None - return rows[0].get("fire_at") if rows else None + return rows[0] if rows else None + + async def pending_wakeup_at(self, session_id: str) -> str | None: + """``fire_at`` of the session's pending wakeup, or ``None``.""" + wakeup = await self.pending_wakeup(session_id) + return wakeup.get("fire_at") if wakeup else None async def _broadcast_session_running( self, session_id: str, is_running: bool, @@ -1595,11 +1613,15 @@ async def _broadcast_session_running( with the transition means the client can redraw the row without a follow-up list fetch — which it skips for the active session. """ + wakeup = await self.pending_wakeup(session_id) or {} await broadcaster.broadcast("__global__", { "type": "session_running", "session_id": session_id, "is_running": is_running, - "pending_wakeup_at": await self.pending_wakeup_at(session_id), + "pending_wakeup_at": wakeup.get("fire_at"), + # Who scheduled the wakeup, and for whom; see GET /api/sessions. + "pending_wakeup_created_by_actor_id": wakeup.get("created_by_actor_id"), + "pending_wakeup_caused_by_actor_id": wakeup.get("caused_by_actor_id"), "has_background_tasks": self._has_live_background_tasks(session_id), }) @@ -3537,6 +3559,8 @@ async def run_cron( run_id: str | None = None, cache_ttl: str = "", effort: str | None = None, + *, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> str: """Run an agent turn for a cron job in an isolated session. @@ -3545,14 +3569,19 @@ async def run_cron( unless the run yielded with a live ``run_in_background`` task, in which case it is kept alive so the agent can resume when the task completes (see ``_teardown_oneshot_client``). + + ``cause``: the cause of the run's session: the system actor (the + default) for a scheduled run; the person who asked for the run when + a person started it. """ if run_id is None: run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") - # A scheduled run is the instance's own work, regardless of who wrote - # the schedule. Human attribution belongs to the schedule mutation. + # A cron run is the instance's own work, regardless of who wrote the + # schedule or who started the run. A person who started it is the + # cause of the run's session. actor = self.db.system_actor session = await self.sessions.create_cron_session( - job_id, run_id=run_id, actor=actor, + job_id, run_id=run_id, actor=actor, cause=cause, ) session_id = session["id"] await self._stamp_cron_session_meta(session_id, "isolated", cache_ttl) @@ -3576,6 +3605,8 @@ async def run_persistent_cron( session_id: str | None = None, cache_ttl: str = "", effort: str | None = None, + *, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> str: """Run a persistent cron job that maintains context across runs. @@ -3591,11 +3622,15 @@ async def run_persistent_cron( which would collide with the parked task — so a persistent-cron background task that outlives its run is not resumed (use an isolated cron for long background work). + + ``cause`` is as for :meth:`run_cron`. It applies only when this call + creates the session; a session of earlier runs keeps its cause. """ session_id = session_id or f"cron:{job_id}" actor = self.db.system_actor await self.sessions.get_or_create( session_id, title=f"Cron: {job_id}", source="cron", actor=actor, + cause=cause, ) await self._stamp_cron_session_meta(session_id, "persistent", cache_ttl) try: diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 5792b0509..2c9f18531 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -625,6 +625,7 @@ async def get_resume_info(self, session_id: str) -> dict | None: async def create_cron_session( self, job_id: str, run_id: str | None = None, *, actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> dict: """Create an isolated session for a single cron run. @@ -633,7 +634,8 @@ async def create_cron_session( ``actor`` is the agent's system principal in production because a scheduled run is the instance's own work, regardless of who wrote the - schedule. + schedule. ``cause`` is the session's cause: by default the actor; the + person for a run that a person started. """ if run_id: session_id = f"cron:{job_id}:{run_id}" @@ -641,6 +643,7 @@ async def create_cron_session( session_id = f"cron:{job_id}" return await self.get_or_create( session_id, title=f"Cron: {job_id}", source="cron", actor=actor, + cause=cause, ) async def create_hook_session( diff --git a/nerve/agent/tools/handlers/wakeups.py b/nerve/agent/tools/handlers/wakeups.py index 63f328752..eba4a5350 100644 --- a/nerve/agent/tools/handlers/wakeups.py +++ b/nerve/agent/tools/handlers/wakeups.py @@ -81,7 +81,10 @@ async def schedule_wakeup_handler(ctx: ToolContext, args: dict) -> ToolResult: is_error=True, ) - wakeup_id = await ctx.engine._record_wakeup(ctx.db, ctx.session_id, args) + # The wakeup turn runs for the cause of the turn that schedules it. + wakeup_id = await ctx.engine._record_wakeup( + ctx.db, ctx.session_id, args, attribution=ctx.attribution, + ) if wakeup_id is None: return ToolResult.text( "No wakeup scheduled: a non-empty `prompt` is required.", diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 70fa0f008..419cd6331 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -27,7 +27,7 @@ load_jobs, ) from nerve.db import Database -from nerve.identity import attribution_for +from nerve.identity import CAUSE_FROM_ACTOR, Attribution, CauseArg, attribution_for if TYPE_CHECKING: from nerve.cron.gates import CronGate @@ -279,6 +279,11 @@ def __init__(self, config: NerveConfig, engine: AgentEngine, db: Database): # interleave scheduler mutations. self._reload_lock = asyncio.Lock() + def _own_attribution(self) -> Attribution: + """Attribution of a run that the instance starts itself: a scheduled + run, a catch-up run, a missed run, or ``nerve cron ``.""" + return attribution_for(self.db.system_actor) + async def start(self) -> None: """Load jobs and start the scheduler.""" # Register drop-in custom gate plugins BEFORE jobs are parsed, so their @@ -394,6 +399,7 @@ async def _record_missed_run(self, job_id: str, due: datetime) -> None: f"run dropped: not started within " f"{_MISFIRE_GRACE_SECONDS}s of its scheduled time" ), + attribution=self._own_attribution(), ) except Exception as e: logger.warning("Failed to log missed run for %s: %s", job_id, e) @@ -903,13 +909,19 @@ async def _current_persistent_session_id(self, job_id: str) -> str | None: return key return None - async def _start_new_generation(self, job_id: str) -> str: - """Create a fresh chat session for a persistent job and map it.""" + async def _start_new_generation( + self, job_id: str, *, cause: CauseArg = CAUSE_FROM_ACTOR, + ) -> str: + """Create a fresh chat session for a persistent job and map it. + + ``cause`` is the session's cause: by default the system actor; the + person whose "Run now" creates the generation. + """ ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") session_id = f"cron:{job_id}:{ts}" await self.engine.sessions.get_or_create( session_id, title=f"Cron: {job_id}", source="cron", - actor=self.db.system_actor, + actor=self.db.system_actor, cause=cause, ) await self.db.set_channel_session(self._channel_key(job_id), session_id) logger.info( @@ -1037,12 +1049,15 @@ def _rotation_reason( return f"age {age_hours:.1f}h >= {rotate_hours}h" return None - async def _resolve_persistent_session(self, job: CronJob) -> tuple[str, bool]: + async def _resolve_persistent_session( + self, job: CronJob, *, cause: CauseArg = CAUSE_FROM_ACTOR, + ) -> tuple[str, bool]: """Pick the chat session for a persistent run, rotating if due. Returns ``(session_id, rotated)``. When rotation is due, the current generation is retired (chat + history preserved as its own session) - and a brand-new chat is minted for this and subsequent runs. + and a brand-new chat is minted for this and subsequent runs, with + ``cause`` as its cause. A current generation keeps its cause. """ current = await self._current_persistent_session_id(job.id) rotated = False @@ -1059,7 +1074,7 @@ async def _resolve_persistent_session(self, job: CronJob) -> tuple[str, bool]: rotated = True if current is None: - current = await self._start_new_generation(job.id) + current = await self._start_new_generation(job.id, cause=cause) return current, rotated # -- End persistent session generations ---------------------------------- @@ -1117,16 +1132,24 @@ def _load_merged_jobs(self, strict: bool = False) -> list[CronJob]: kept, self._rejected_job_ids = _drop_reserved(list(jobs_by_id.values())) return kept - async def _run_job_wrapper(self, job: CronJob) -> None: - """Wrapper to run a cron job with logging and optional lock.""" + async def _run_job_wrapper( + self, job: CronJob, attribution: Attribution | None = None, + ) -> None: + """Wrapper to run a cron job with logging and optional lock. + + ``attribution`` names who started the run and its cause. ``None`` + (the scheduler's call) is the instance's own run. + """ if job.lock: lock = self._job_locks.setdefault(job.id, asyncio.Lock()) async with lock: - await self._run_job_inner(job) + await self._run_job_inner(job, attribution) else: - await self._run_job_inner(job) + await self._run_job_inner(job, attribution) - async def _run_job_inner(self, job: CronJob) -> None: + async def _run_job_inner( + self, job: CronJob, attribution: Attribution | None = None, + ) -> None: """Inner implementation of job execution.""" # Pre-check: skip if any configured run gate is unsatisfied. if job.gates: @@ -1139,7 +1162,8 @@ async def _run_job_inner(self, job: CronJob) -> None: logger.info("Skipping cron job %s: %s", job.id, decision.reason) return - log_id = await self.db.log_cron_start(job.id) + attribution = attribution or self._own_attribution() + log_id = await self.db.log_cron_start(job.id, attribution=attribution) logger.info("Running cron job: %s (mode=%s)", job.id, job.session_mode) if job.workflow is not None: @@ -1164,7 +1188,9 @@ async def _run_job_inner(self, job: CronJob) -> None: if job.session_mode == "persistent": # Resolve the current generation chat, rotating to a fresh # one first when due (the old chat is preserved). - session_id, rotated = await self._resolve_persistent_session(job) + session_id, rotated = await self._resolve_persistent_session( + job, cause=attribution.caused_by_actor_id, + ) else: # Isolated mode: per-run session. The run_id is generated # here (the engine would otherwise generate an identical @@ -1211,6 +1237,7 @@ async def _run_job_inner(self, job: CronJob) -> None: session_id=session_id, cache_ttl=job.cache_ttl, effort=effort, + cause=attribution.caused_by_actor_id, ) else: response = await self.engine.run_cron( @@ -1220,6 +1247,7 @@ async def _run_job_inner(self, job: CronJob) -> None: run_id=run_id, cache_ttl=job.cache_ttl, effort=effort, + cause=attribution.caused_by_actor_id, ) # Keep the tail of the response — for multi-message runs the @@ -1316,9 +1344,16 @@ async def _start_workflow_run(self, job: CronJob, log_id: int) -> None: ) await self.db.log_cron_finish(log_id, "error", error=str(e)) - async def _run_source_wrapper(self, runner: SourceRunner) -> None: - """Wrapper to run a source ingestion with cron and source logging.""" - log_id = await self.db.log_cron_start(runner.job_id) + async def _run_source_wrapper( + self, runner: SourceRunner, attribution: Attribution | None = None, + ) -> None: + """Wrapper to run a source ingestion with cron and source logging. + + ``attribution`` is as for :meth:`_run_job_wrapper`. + """ + log_id = await self.db.log_cron_start( + runner.job_id, attribution=attribution or self._own_attribution(), + ) logger.info("Running source: %s", runner.source.source_name) try: @@ -1410,7 +1445,10 @@ def _dispatch_wakeup(self, session_id: str, wakeup: dict) -> None: # The actor is the instance's own: a model-scheduled wakeup, or a # deferral this service is delivering. Who wrote a run-later message # is already recorded on the row the route persisted when they - # composed it. + # composed it. The cause is the one stored when the wakeup was + # scheduled: the person who deferred the message, or the cause of + # the turn that called ScheduleWakeup. A stored NULL (the cause was + # unknown, or the row is older than v057) stays unknown. task = asyncio.create_task( self.engine.run( session_id=session_id, @@ -1418,6 +1456,7 @@ def _dispatch_wakeup(self, session_id: str, wakeup: dict) -> None: source=source, internal=True, actor=self.db.system_actor, + cause=wakeup.get("caused_by_actor_id"), # A run-later deferral is a message the user wrote and asked # to be delivered now, so it belongs back at the top of the # sidebar. A model's own ScheduleWakeup tick is the session @@ -1436,8 +1475,15 @@ def _done(t: asyncio.Task) -> None: task.add_done_callback(_done) - async def run_job(self, job_id: str) -> None: - """Run a specific job manually (used by CLI).""" + async def run_job( + self, job_id: str, *, attribution: Attribution | None = None, + ) -> None: + """Run a specific job manually (the CLI and the REST trigger). + + ``attribution``: the person who asked for the run through the API, as + actor and cause. The run's turn has the system actor as actor and + this cause. ``None`` is the instance's own run (``nerve cron ``). + """ job = next((j for j in self._jobs if j.id == job_id), None) if not job: # Try loading fresh from both files @@ -1447,7 +1493,7 @@ async def run_job(self, job_id: str) -> None: if not job: raise ValueError(f"Job not found: {job_id}") - await self._run_job_wrapper(job) + await self._run_job_wrapper(job, attribution) async def rotate_session(self, job_id: str) -> dict: """Force-rotate a persistent cron to a fresh chat session. diff --git a/nerve/db/actor_columns.py b/nerve/db/actor_columns.py index 3a813fc35..39b4d58c6 100644 --- a/nerve/db/actor_columns.py +++ b/nerve/db/actor_columns.py @@ -41,4 +41,8 @@ ("review_loops", "killed_caused_by_actor_id"), ("review_loop_decisions", "actor_id"), ("review_loop_decisions", "caused_by_actor_id"), + ("session_wakeups", "created_by_actor_id"), + ("session_wakeups", "caused_by_actor_id"), + ("cron_logs", "actor_id"), + ("cron_logs", "caused_by_actor_id"), ) diff --git a/nerve/db/cron.py b/nerve/db/cron.py index 82090e7a3..2ac8ec57d 100644 --- a/nerve/db/cron.py +++ b/nerve/db/cron.py @@ -2,15 +2,25 @@ from __future__ import annotations +from nerve.identity import UNATTRIBUTED, Attribution from nerve.utils.time import utc_now_iso class CronStore: - """Mixin providing cron job logging operations.""" + """Mixin providing cron job logging operations. - async def log_cron_start(self, job_id: str) -> int: + Each run row records who started the run (``actor_id``) and on whose + behalf it happens (``caused_by_actor_id``): the system actor for a + scheduled run, the person for a run that they started through the API. + """ + + async def log_cron_start( + self, job_id: str, *, attribution: Attribution = UNATTRIBUTED, + ) -> int: result = await self._write( - "INSERT INTO cron_logs (job_id) VALUES (?)", (job_id,) + "INSERT INTO cron_logs (job_id, actor_id, caused_by_actor_id) " + "VALUES (?, ?, ?)", + (job_id, attribution.actor_id, attribution.caused_by_actor_id), ) return result.lastrowid @@ -42,6 +52,7 @@ async def log_cron_finish( async def log_cron_missed( self, job_id: str, scheduled_run_time: str, error: str | None = None, + *, attribution: Attribution = UNATTRIBUTED, ) -> int: """Record a run the scheduler dropped for starting too late. @@ -53,9 +64,12 @@ async def log_cron_missed( """ now = utc_now_iso() result = await self._write( - "INSERT INTO cron_logs (job_id, started_at, finished_at, status, error) " - "VALUES (?, ?, ?, 'missed', ?)", - (job_id, scheduled_run_time, now, error), + "INSERT INTO cron_logs (job_id, started_at, finished_at, status, error, " + "actor_id, caused_by_actor_id) VALUES (?, ?, ?, 'missed', ?, ?, ?)", + ( + job_id, scheduled_run_time, now, error, + attribution.actor_id, attribution.caused_by_actor_id, + ), ) return result.lastrowid diff --git a/nerve/db/migrations/v057_schedule_attribution.py b/nerve/db/migrations/v057_schedule_attribution.py new file mode 100644 index 000000000..01d709e6b --- /dev/null +++ b/nerve/db/migrations/v057_schedule_attribution.py @@ -0,0 +1,62 @@ +"""V57: record who schedules a wakeup and who starts a cron run. + +Adds four nullable columns that hold an ``actor_refs.id``: + +- ``session_wakeups.created_by_actor_id``: who scheduled the wakeup. A person + who deferred a message with "Run later", or the system actor when the agent + called ``ScheduleWakeup`` / ``schedule_wakeup`` in a turn. +- ``session_wakeups.caused_by_actor_id``: on whose behalf the wakeup turn + runs. The person for "Run later", the cause of the turn that called the + tool for an agent wakeup. The turn that the wakeup fires runs for this + cause. +- ``cron_logs.actor_id``: who started the run. The system actor for a + scheduled run, a catch-up run or a missed run; the person who used "Run + now" through the REST API. +- ``cron_logs.caused_by_actor_id``: on whose behalf the run happens. The + system actor for scheduled work, the person for "Run now". + +**Existing rows stay NULL and nothing is backfilled.** NULL reads as "not +recorded". A wakeup row from before this migration fires with an unknown +cause. + +The foreign keys follow v048: an ID that resolves to no actor can never be +stored, and NULL is exempt from the check. Each column is added only when the +table does not have it yet, so a second run does nothing. +""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +# (table, column, DDL type + constraints) +_COLUMNS = ( + ("session_wakeups", "created_by_actor_id", "TEXT REFERENCES actor_refs(id)"), + ("session_wakeups", "caused_by_actor_id", "TEXT REFERENCES actor_refs(id)"), + ("cron_logs", "actor_id", "TEXT REFERENCES actor_refs(id)"), + ("cron_logs", "caused_by_actor_id", "TEXT REFERENCES actor_refs(id)"), +) + + +async def _has_column(db: aiosqlite.Connection, table: str, column: str) -> bool: + async with db.execute(f"PRAGMA table_info({table})") as cursor: + rows = await cursor.fetchall() + return any(row[1] == column for row in rows) + + +async def up(db: aiosqlite.Connection) -> None: + added = [] + for table, column, ddl in _COLUMNS: + if await _has_column(db, table, column): + continue + # SQLite allows a REFERENCES clause on ADD COLUMN when the default is + # NULL, so the table is not rebuilt and every existing row is NULL. + await db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}") + added.append(f"{table}.{column}") + logger.info( + "v057: schedule attribution columns %s", + ", ".join(added) if added else "already present", + ) diff --git a/nerve/db/wakeups.py b/nerve/db/wakeups.py index 645aa141e..e90c74c48 100644 --- a/nerve/db/wakeups.py +++ b/nerve/db/wakeups.py @@ -4,10 +4,16 @@ the ``ScheduleWakeup`` tool. Rows are written by the PostToolUse capture hook and consumed by the cron-service sweep, which fires the prompt through ``engine.run(..., source="wakeup")`` at ``fire_at``. + +Each row records who scheduled it (``created_by_actor_id``) and on whose +behalf it runs (``caused_by_actor_id``). The turn that the wakeup fires runs +for that cause. """ from __future__ import annotations +from nerve.identity import UNATTRIBUTED, Attribution + class WakeupStore: """Mixin providing session wakeup persistence.""" @@ -18,6 +24,8 @@ async def add_wakeup( prompt: str, fire_at: str, reason: str = "", + *, + attribution: Attribution = UNATTRIBUTED, ) -> int: """Record a pending wakeup, replacing any prior pending one. @@ -25,6 +33,9 @@ async def add_wakeup( ``/loop`` self-pacing model re-calls the tool each turn). So a new request supersedes any earlier pending wakeup for the session. + ``attribution`` names who scheduled the wakeup and the cause that + its turn runs for. + Returns the new wakeup id. """ async with self._atomic(): @@ -34,9 +45,14 @@ async def add_wakeup( (session_id,), ) cursor = await self.db.execute( - """INSERT INTO session_wakeups (session_id, prompt, reason, fire_at) - VALUES (?, ?, ?, ?)""", - (session_id, prompt, reason, fire_at), + """INSERT INTO session_wakeups + (session_id, prompt, reason, fire_at, + created_by_actor_id, caused_by_actor_id) + VALUES (?, ?, ?, ?, ?, ?)""", + ( + session_id, prompt, reason, fire_at, + attribution.actor_id, attribution.caused_by_actor_id, + ), ) wakeup_id = cursor.lastrowid await cursor.close() diff --git a/nerve/gateway/routes/cron.py b/nerve/gateway/routes/cron.py index e9be2a044..b1a910eb3 100644 --- a/nerve/gateway/routes/cron.py +++ b/nerve/gateway/routes/cron.py @@ -6,6 +6,8 @@ from nerve.gateway.auth import require_auth from nerve.gateway.routes._deps import get_deps +from nerve.identity import Actor, attribution_for + router = APIRouter(dependencies=[Depends(require_auth)]) @@ -38,23 +40,28 @@ async def reload_cron_jobs(): @router.post("/api/cron/jobs/{job_id}/trigger") -async def trigger_cron_job(job_id: str): - """Manually trigger a specific cron job or source runner.""" +async def trigger_cron_job(job_id: str, actor: Actor = Depends(require_auth)): + """Manually trigger a specific cron job or source runner. + + The run log records the person as actor and cause. A job's turn runs as + the system actor on their behalf. + """ from nerve.gateway.server import _cron_service if not _cron_service: raise HTTPException(status_code=503, detail="Cron service not available") + attribution = attribution_for(actor) # Source runner runners = getattr(_cron_service, "_source_runners", []) runner = next((r for r in runners if r.job_id == job_id), None) if runner: - await _cron_service._run_source_wrapper(runner) + await _cron_service._run_source_wrapper(runner, attribution) return {"job_id": job_id, "triggered": True} # Regular cron job try: - await _cron_service.run_job(job_id) + await _cron_service.run_job(job_id, attribution=attribution) return {"job_id": job_id, "triggered": True} except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) diff --git a/nerve/gateway/routes/sessions.py b/nerve/gateway/routes/sessions.py index 80842f592..d35491396 100644 --- a/nerve/gateway/routes/sessions.py +++ b/nerve/gateway/routes/sessions.py @@ -161,8 +161,8 @@ def _page_size() -> int | None: return size if size and size > 0 else None -async def _pending_wakeup_map(deps) -> dict[str, str]: - """``session_id -> fire_at`` for every pending wakeup, in one query. +async def _pending_wakeup_map(deps) -> dict[str, dict]: + """``session_id -> wakeup row`` for every pending wakeup, in one query. ``ScheduleWakeup`` keeps at most one pending row per session, so a plain map is enough. Enrichment: a failure dims the indicator, it must never @@ -172,7 +172,7 @@ async def _pending_wakeup_map(deps) -> dict[str, str]: rows = await deps.db.list_pending_wakeups() except Exception: # noqa: BLE001 — enrichment must never break listing return {} - return {r["session_id"]: r["fire_at"] for r in rows if r.get("session_id")} + return {r["session_id"]: r for r in rows if r.get("session_id")} async def _decorate(deps, sessions: list[dict]) -> list[dict]: @@ -183,11 +183,16 @@ async def _decorate(deps, sessions: list[dict]) -> list[dict]: # a wakeup is scheduled, or a background task is still running inside its # CLI. The sidebar keeps those parked sessions in the "Running" group # (with their own dot colour) instead of dropping them to idle. - wakeup_at = await _pending_wakeup_map(deps) + wakeups = await _pending_wakeup_map(deps) for s in sessions: s["is_running"] = s["id"] in running_ids s["awaiting_input"] = s["id"] in awaiting_ids - s["pending_wakeup_at"] = wakeup_at.get(s["id"]) + wakeup = wakeups.get(s["id"]) or {} + s["pending_wakeup_at"] = wakeup.get("fire_at") + # Who scheduled the wakeup (a person, or the system actor for the + # agent) and on whose behalf it runs. + s["pending_wakeup_created_by_actor_id"] = wakeup.get("created_by_actor_id") + s["pending_wakeup_caused_by_actor_id"] = wakeup.get("caused_by_actor_id") s["has_background_tasks"] = deps.engine.has_live_background_tasks(s["id"]) await _attach_review_loops(deps, sessions) return sessions @@ -717,8 +722,11 @@ async def run_later(req: RunLaterRequest, actor: Actor = Depends(require_auth)): fire_at = ( datetime.now(timezone.utc) + timedelta(seconds=seconds) ).isoformat() + # The person schedules the deferred turn, and it runs on their + # behalf when it fires. await deps.db.add_wakeup( req.session_id, prompt=message, fire_at=fire_at, reason="run-later", + attribution=attribution_for(actor), ) else: ack = _RUN_LATER_MANUAL_ACK diff --git a/tests/test_cron.py b/tests/test_cron.py index a95596fc0..ba88803c3 100644 --- a/tests/test_cron.py +++ b/tests/test_cron.py @@ -502,7 +502,10 @@ async def test_fires_overdue_jobs(self, cron_service): await cron_service._catchup_missed_jobs() - cron_service.db.log_cron_start.assert_called_once_with("overdue-job") + # A catch-up run is the instance's own work. + cron_service.db.log_cron_start.assert_called_once_with( + "overdue-job", attribution=attribution_for(FAKE_SYSTEM_ACTOR), + ) cron_service.engine.run_cron.assert_called_once() @pytest.mark.asyncio @@ -791,9 +794,10 @@ async def test_rotation_starts_new_chat_and_preserves_old(self, cron_service): # New chat is created and becomes the job's current session. cron_service.engine.sessions.get_or_create.assert_awaited_once_with( session_id, title="Cron: pers", source="cron", - # Who a cron generation belongs to has its own test - # (test_attribution.py); this one is about rotation. - actor=ANY, + # Who a cron generation belongs to, and its cause, have their + # own tests (test_attribution.py, test_schedule_attribution.py); + # this one is about rotation. + actor=ANY, cause=ANY, ) cron_service.db.set_channel_session.assert_awaited_once_with( "cron:pers", session_id, @@ -1061,7 +1065,9 @@ async def test_runs_when_tasks_gate_satisfied(self, cron_service): await cron_service._run_job_inner(job) - cron_service.db.log_cron_start.assert_called_once_with("planner") + cron_service.db.log_cron_start.assert_called_once_with( + "planner", attribution=attribution_for(FAKE_SYSTEM_ACTOR), + ) cron_service.engine.run_cron.assert_called_once() @pytest.mark.asyncio diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py new file mode 100644 index 000000000..c42aaa25f --- /dev/null +++ b/tests/test_schedule_attribution.py @@ -0,0 +1,703 @@ +"""Schedules: who schedules a wakeup or starts a cron run, and for whom. + +A wakeup row records who scheduled it and its cause: a person who defers a +message with "Run later", or the system actor and the session's cause when +the agent calls ``ScheduleWakeup`` / ``schedule_wakeup``. A wakeup fires in +its session, so its turn runs for the session's cause. A cron run log records +who started the run: the system actor for a scheduled run, the person who +used "Run now". A run that a person started creates its session with that +person as the cause. + +The tests drive the real routes, cron service, engine, tool registry and +database. The model runtime is the tool-calling stand-in of +``tests/test_cause_attribution.py``, or a stand-in for the Claude CLI process +where the CLI's own ScheduleWakeup hook is under test. +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + +import pytest +from claude_agent_sdk import AssistantMessage, ResultMessage, TextBlock + +from nerve.agent.backends import claude as claude_backend +from nerve.agent.backends.claude import translate_message +from nerve.cron.jobs import CronJob +from nerve.cron.service import CronService +from nerve.db.actor_columns import ACTOR_COLUMNS +from nerve.identity import Attribution +from tests.test_cause_attribution import ( # noqa: F401 — fixtures + ALICE_PRINCIPAL, + BOB_PRINCIPAL, + _as, + _chat, + _client, + _Install, + _new_session, + external_install, + install, +) + +_SCHEDULE_COLUMNS = ( + ("session_wakeups", "created_by_actor_id"), + ("session_wakeups", "caused_by_actor_id"), + ("cron_logs", "actor_id"), + ("cron_logs", "caused_by_actor_id"), +) + + +def _past() -> str: + return (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat() + + +async def _wakeups(install: _Install, session_id: str) -> list[tuple]: + async with install.db.db.execute( + "SELECT prompt, status, created_by_actor_id, caused_by_actor_id" + " FROM session_wakeups WHERE session_id = ? ORDER BY id", + (session_id,), + ) as cursor: + return [tuple(row) async for row in cursor] + + +async def _cron_logs(install: _Install) -> list[tuple]: + async with install.db.db.execute( + "SELECT job_id, status, actor_id, caused_by_actor_id FROM cron_logs ORDER BY id" + ) as cursor: + return [tuple(row) async for row in cursor] + + +async def _fire_due_wakeups(install: _Install) -> CronService: + """Make every pending wakeup due, run one sweep and wait for its turns.""" + await install.db._write( + "UPDATE session_wakeups SET fire_at = ? WHERE status = 'pending'", (_past(),), + ) + service = CronService(install.config, install.engine, install.db) + before = asyncio.all_tasks() + await service._sweep_wakeups() + fired = asyncio.all_tasks() - before + assert fired, "the sweep fired no wakeup" + await asyncio.wait_for(asyncio.gather(*fired), timeout=10) + return service + + +def _schedule_in_turn(install: _Install, session_id: str, prompt: str): + """A tool call that the agent makes in the session's current turn.""" + + async def _call() -> None: + engine = install.engine + result = await engine.registry.invoke( + "schedule_wakeup", engine._build_tool_context(session_id), + {"delaySeconds": 120, "prompt": prompt}, + ) + assert not result.is_error, result + + return _call + + +# --------------------------------------------------------------------------- # +# Schema # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_v057_keeps_old_rows_null_and_is_idempotent(install): + raw = install.db.db + for table, column in _SCHEDULE_COLUMNS: + await raw.execute(f"ALTER TABLE {table} DROP COLUMN {column}") + await raw.execute( + "INSERT INTO sessions (id, title, source, status, created_at, updated_at)" + " VALUES ('old', 'Old chat', 'web', 'idle', '2020-01-01', '2020-01-01')" + ) + await raw.execute( + "INSERT INTO session_wakeups (session_id, prompt, fire_at)" + " VALUES ('old', 'from before', '2020-01-01T00:00:00+00:00')" + ) + await raw.execute("INSERT INTO cron_logs (job_id) VALUES ('old-job')") + await raw.commit() + + from nerve.db.migrations.v057_schedule_attribution import up + + await up(raw) + await up(raw) + await raw.commit() + assert await _wakeups(install, "old") == [("from before", "pending", None, None)] + assert await _cron_logs(install) == [("old-job", None, None, None)] + assert set(_SCHEDULE_COLUMNS) <= set(ACTOR_COLUMNS) + + +# --------------------------------------------------------------------------- # +# Run later # +# --------------------------------------------------------------------------- # + + +async def _run_later(client, headers, session_id: str, message: str) -> None: + response = await client.post( + "/api/sessions/run-later", headers=headers, + json={"session_id": session_id, "message": message, "delay": "30m"}, + ) + assert response.status_code == 200, response.text + assert response.json()["scheduled"] is True + + +@pytest.mark.asyncio +async def test_a_run_later_fires_as_the_system_actor_for_the_person( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + alice = install.actor_id("alice") + system = install.system_actor_id + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _run_later(client, install.headers("alice"), session_id, "summarize the week") + + assert await _wakeups(install, session_id) == [ + ("summarize the week", "pending", alice, alice), + ] + + await _fire_due_wakeups(install) + + turn = install.turn_in(session_id) + assert turn["cause"] == alice + assert await install.events(turn["task_id"]) == [ + (None, "pending", session_id, system, alice), + ] + assert await install.session_cause_of(session_id) == alice + # The person's own message is the only user row; the delivery is internal. + assert await install.user_rows(session_id) == [("summarize the week", alice)] + assert await _wakeups(install, session_id) == [ + ("summarize the week", "fired", alice, alice), + ] + + +@pytest.mark.asyncio +async def test_a_run_later_in_another_persons_session_fires_for_the_session_cause( + install, monkeypatch, +): + """Bob defers a message in Alice's session. + + The wakeup row names Bob, who scheduled it. It fires in Alice's session, + whose turns run for Alice (the session cause). + """ + install.use_fake_runtime(monkeypatch) + alice, bob = install.actor_id("alice"), install.actor_id("bob") + system = install.system_actor_id + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _run_later(client, install.headers("bob"), session_id, "deferred work") + + assert await _wakeups(install, session_id) == [ + ("deferred work", "pending", bob, bob), + ] + + await _fire_due_wakeups(install) + + fired = install.turn_in(session_id) + assert fired["cause"] == alice + assert await install.events(fired["task_id"]) == [ + (None, "pending", session_id, system, alice), + ] + + +# --------------------------------------------------------------------------- # +# Wakeups that the agent sets # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_a_wakeup_that_the_agent_sets_runs_for_the_sessions_cause( + install, monkeypatch, +): + """Bob's session: the agent sets a wakeup in a turn that Alice sends. + + The wakeup records the system actor and Bob, the session cause, and it + fires for Bob. + """ + install.use_fake_runtime(monkeypatch) + bob = install.actor_id("bob") + system = install.system_actor_id + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("bob")) + runtime = install.runtime(session_id) + runtime.before_tool_call = _schedule_in_turn(install, session_id, "check the build") + await _chat(client, install.headers("alice"), session_id, "watch the build") + runtime.before_tool_call = None + + assert await _wakeups(install, session_id) == [ + ("check the build", "pending", system, bob), + ] + + await _fire_due_wakeups(install) + + fired = install.turns_in(session_id)[-1] + assert fired["cause"] == bob + assert await install.events(fired["task_id"]) == [ + (None, "pending", session_id, system, bob), + ] + + +@pytest.mark.asyncio +async def test_a_wakeup_set_in_a_turn_with_no_cause_fires_with_no_cause( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + system = install.system_actor_id + await install.db.create_session("chat-app", source="telegram", actor=None) + install.runtime("chat-app").before_tool_call = _schedule_in_turn( + install, "chat-app", "ask again later", + ) + await install.engine.run( + "chat-app", "remind me", source="telegram", actor=None, + ) + install.runtime("chat-app").before_tool_call = None + assert await _wakeups(install, "chat-app") == [ + ("ask again later", "pending", system, None), + ] + + await _fire_due_wakeups(install) + + fired = install.turns_in("chat-app")[-1] + assert fired["cause"] is None + assert await install.events(fired["task_id"]) == [ + (None, "pending", "chat-app", system, None), + ] + assert await install.session_cause_of("chat-app") is None + + +class _FakeCli: + """Stands in for one Claude CLI process (``ClaudeClient``). + + A turn calls the ScheduleWakeup PostToolUse hook from the options that + the backend built, when ``schedule`` holds a prompt, as the CLI does after + its built-in tool runs. Then it makes a tool call through the session's + tool context and completes. + """ + + started: list[_FakeCli] = [] + install: _Install | None = None + + def __init__(self, spec, options): + self.spec = spec + self.options = options + self.schedule: str | None = None + self.started.append(self) + + @property + def native_session_id(self): + return "native-cli" + + async def connect(self): + pass + + async def start_turn(self, turn): + pass + + async def receive_turn(self): + if self.schedule: + tool_input = {"delaySeconds": 300, "prompt": self.schedule} + self.schedule = None + for matcher in self.options.hooks["PostToolUse"]: + if matcher.matcher == "ScheduleWakeup": + for hook in matcher.hooks: + await hook( + {"tool_name": "ScheduleWakeup", "tool_input": tool_input}, + "toolu-1", {}, + ) + await self.install.agent_tool_call(self.spec.session_id) + for msg in ( + AssistantMessage(content=[TextBlock(text="ok")], model="claude-test"), + ResultMessage( + subtype="success", duration_ms=1, duration_api_ms=1, + is_error=False, num_turns=1, session_id="native-cli", + total_cost_usd=0.0, usage={"input_tokens": 1}, + ), + ): + for event in translate_message(msg): + yield event + + async def interrupt(self): + pass + + def is_alive(self): + return True + + async def disconnect(self, timeout: float = 5.0): + pass + + def try_receive_idle_events(self): + return None + + async def receive_idle_events(self, timeout): + await asyncio.sleep(min(timeout or 0, 0.05)) + raise asyncio.TimeoutError + + def buffer_used(self): + return 0 + + +@pytest.fixture +def cli(monkeypatch) -> list[_FakeCli]: + started: list[_FakeCli] = [] + monkeypatch.setattr(_FakeCli, "started", started) + monkeypatch.setattr(claude_backend, "ClaudeClient", _FakeCli) + monkeypatch.setattr( + claude_backend.ClaudeBackend, "validate_resume_target", + lambda self, native_id, cwd: True, + ) + return started + + +async def _stop_idle_watchers(install: _Install) -> None: + for session_id in list(install.engine._idle_watchers): + install.engine._stop_idle_watcher(session_id) + + +@pytest.mark.asyncio +async def test_a_cli_wakeup_records_the_sessions_cause( + install, cli, monkeypatch, +): + """The Claude CLI's built-in ScheduleWakeup, captured by its hook. + + Bob sends the turn in Alice's session; the wakeup and its turn are + Alice's, the session cause. + """ + monkeypatch.setattr(_FakeCli, "install", install) + alice = install.actor_id("alice") + system = install.system_actor_id + try: + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _chat(client, install.headers("alice"), session_id, "hello") + (process,) = cli + process.schedule = "look at the queue again" + await _chat(client, install.headers("bob"), session_id, "watch the queue") + + assert await _wakeups(install, session_id) == [ + ("look at the queue again", "pending", system, alice), + ] + + await _fire_due_wakeups(install) + + assert [turn["cause"] for turn in install.turns_in(session_id)] == [ + alice, alice, alice, + ] + finally: + await _stop_idle_watchers(install) + + +# --------------------------------------------------------------------------- # +# Cron runs # +# --------------------------------------------------------------------------- # + + +def _cron_service(install: _Install, *jobs: CronJob) -> CronService: + service = CronService(install.config, install.engine, install.db) + service._jobs = list(jobs) + return service + + +@pytest.mark.asyncio +async def test_a_scheduled_cron_run_is_the_systems_for_the_system(install, monkeypatch): + install.use_fake_runtime(monkeypatch) + system = install.system_actor_id + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + service = _cron_service(install, job) + + # The scheduler calls the wrapper with the job only. + await service._run_job_wrapper(job) + + assert await _cron_logs(install) == [("nightly", "success", system, system)] + turn = install.turns[0] + assert turn["cause"] == system + assert await install.user_rows(turn["session_id"]) == [("tidy the board", system)] + assert await install.session_cause_of(turn["session_id"]) == system + assert await install.events(turn["task_id"]) == [ + (None, "pending", turn["session_id"], system, system), + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("session_mode", ["isolated", "persistent"]) +async def test_run_now_through_rest_runs_as_the_system_for_the_person( + install, monkeypatch, session_mode, +): + install.use_fake_runtime(monkeypatch) + alice = install.actor_id("alice") + system = install.system_actor_id + job = CronJob( + id="nightly", schedule="1h", prompt="tidy the board", session_mode=session_mode, + ) + monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) + + async with _client(install.app()) as client: + response = await client.post( + "/api/cron/jobs/nightly/trigger", headers=install.headers("alice"), + ) + assert response.status_code == 200, response.text + logs = (await client.get( + "/api/cron/logs", headers=install.headers("bob"), + )).json()["logs"] + + assert [(log["actor_id"], log["caused_by_actor_id"]) for log in logs] == [ + (alice, alice), + ] + turn = install.turns[0] + assert turn["cause"] == alice + assert await install.user_rows(turn["session_id"]) == [("tidy the board", system)] + assert await install.events(turn["task_id"]) == [ + (None, "pending", turn["session_id"], system, alice), + ] + assert await install.session_cause_of(turn["session_id"]) == alice + + +@pytest.mark.asyncio +async def test_run_now_of_a_source_records_the_person(install, monkeypatch): + bob = install.actor_id("bob") + + async def _run(): + return SimpleNamespace(records_ingested=2, records_dropped=0, error=None) + + runner = SimpleNamespace( + job_id="source:fake", source=SimpleNamespace(source_name="fake"), run=_run, + ) + service = _cron_service(install) + service._source_runners = [runner] + monkeypatch.setattr("nerve.gateway.server._cron_service", service) + + async with _client(install.app()) as client: + response = await client.post( + "/api/cron/jobs/source:fake/trigger", headers=install.headers("bob"), + ) + assert response.status_code == 200, response.text + + # A scheduled run of the same source. + await service._run_source_wrapper(runner) + + system = install.system_actor_id + assert await _cron_logs(install) == [ + ("source:fake", "success", bob, bob), + ("source:fake", "success", system, system), + ] + + +@pytest.mark.asyncio +async def test_a_missed_run_is_the_systems(install): + service = _cron_service(install) + await service._record_missed_run("nightly", datetime.now(timezone.utc)) + + system = install.system_actor_id + assert await _cron_logs(install) == [("nightly", "missed", system, system)] + + +@pytest.mark.asyncio +async def test_the_cli_run_of_a_job_is_the_systems(install, monkeypatch): + """``nerve cron `` runs the job with no attribution argument.""" + install.use_fake_runtime(monkeypatch) + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + await _cron_service(install, job).run_job("nightly") + + system = install.system_actor_id + assert await _cron_logs(install) == [("nightly", "success", system, system)] + assert install.turns[0]["cause"] == system + + +# --------------------------------------------------------------------------- # +# Payloads # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_the_session_list_names_who_scheduled_the_pending_wakeup( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + alice, bob = install.actor_id("alice"), install.actor_id("bob") + system = install.system_actor_id + async with _client(install.app()) as client: + deferred = await _new_session(client, install.headers("alice")) + await _run_later(client, install.headers("alice"), deferred, "later") + looping = await _new_session(client, install.headers("alice")) + install.runtime(looping).before_tool_call = _schedule_in_turn( + install, looping, "next tick", + ) + await _chat(client, install.headers("bob"), looping, "loop on this") + plain = await _new_session(client, install.headers("bob")) + rows = { + row["id"]: row + for row in ( + await client.get("/api/sessions", headers=install.headers("bob")) + ).json()["sessions"] + } + + def _scheduled_by(session_id: str) -> tuple: + row = rows[session_id] + return ( + row["pending_wakeup_at"] is not None, + row["pending_wakeup_created_by_actor_id"], + row["pending_wakeup_caused_by_actor_id"], + ) + + assert _scheduled_by(deferred) == (True, alice, alice) + # The agent set it in a turn that Bob sent; Alice is the session cause. + assert _scheduled_by(looping) == (True, system, alice) + assert _scheduled_by(plain) == (False, None, None) + + +@pytest.mark.asyncio +async def test_the_session_running_event_names_who_scheduled_the_wakeup( + install, monkeypatch, +): + from nerve.agent import engine as engine_module + + bob = install.actor_id("bob") + await install.db.create_session("s1", source="web", actor=None) + await install.db.add_wakeup( + "s1", prompt="tick", fire_at=_past(), + attribution=Attribution(install.system_actor_id, bob), + ) + events: list[dict] = [] + + async def _broadcast(channel, message): + events.append(message) + + monkeypatch.setattr(engine_module.broadcaster, "broadcast", _broadcast) + await install.engine._broadcast_session_running("s1", False) + + (event,) = events + assert event["type"] == "session_running" + assert event["pending_wakeup_at"] is not None + assert event["pending_wakeup_created_by_actor_id"] == install.system_actor_id + assert event["pending_wakeup_caused_by_actor_id"] == bob + + +@pytest.mark.asyncio +async def test_the_actor_directory_names_every_scheduler_and_cause( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _run_later(client, install.headers("alice"), session_id, "later") + await client.post("/api/cron/jobs/nightly/trigger", headers=install.headers("bob")) + directory = { + row["id"] + for row in ( + await client.get("/api/actors", headers=install.headers("alice")) + ).json()["actors"] + } + + for table, column in _SCHEDULE_COLUMNS: + async with install.db.db.execute( + f"SELECT DISTINCT {column} FROM {table} WHERE {column} IS NOT NULL" + ) as cursor: + referenced = {row[0] async for row in cursor} + assert referenced, (table, column) + assert referenced <= directory, (table, column) + + +# --------------------------------------------------------------------------- # +# External mode # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_external_mode_run_later_and_run_now_name_the_principal( + external_install, monkeypatch, +): + install = external_install + install.use_fake_runtime(monkeypatch) + alice = _as(ALICE_PRINCIPAL, "Alice") + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) + async with _client(install.app()) as client: + session_id = await _new_session(client, alice) + await _run_later(client, alice, session_id, "summarize the week") + await client.post("/api/cron/jobs/nightly/trigger", headers=alice) + + assert await _wakeups(install, session_id) == [ + ("summarize the week", "pending", ALICE_PRINCIPAL, ALICE_PRINCIPAL), + ] + assert await _cron_logs(install) == [ + ("nightly", "success", ALICE_PRINCIPAL, ALICE_PRINCIPAL), + ] + + await _fire_due_wakeups(install) + + assert install.turn_in(session_id)["cause"] == ALICE_PRINCIPAL + assert {turn["cause"] for turn in install.turns} == {ALICE_PRINCIPAL} + + +def _named_principals(process: _FakeCli) -> list[str]: + """The principal headers that the CLI process sends on model requests.""" + from nerve.inference_principal import CUSTOM_HEADERS_ENV, INFERENCE_PRINCIPAL_HEADER + + named = [] + for line in process.options.env.get(CUSTOM_HEADERS_ENV, "").splitlines(): + name, _, value = line.partition(":") + if name.strip().lower() == INFERENCE_PRINCIPAL_HEADER.lower(): + named.append(value.strip()) + return named + + +@pytest.mark.asyncio +async def test_external_mode_wakeup_turns_name_the_session_cause_to_the_model_api( + external_install, cli, monkeypatch, +): + """A fired wakeup in Alice's session names Alice on its model requests. + + Bob sends the turn in which the agent sets a wakeup, and Bob defers a + message in another of Alice's sessions. Both fire in Alice's sessions, + whose CLI processes name Alice, the session cause. + """ + pytest.importorskip("nerve.inference_principal") + install = external_install + monkeypatch.setattr(_FakeCli, "install", install) + monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False) + alice, bob = _as(ALICE_PRINCIPAL, "Alice"), _as(BOB_PRINCIPAL, "Bob") + try: + async with _client(install.app()) as client: + looping = await _new_session(client, alice) + await _chat(client, alice, looping, "hello") + cli[-1].schedule = "check the build" + await _chat(client, bob, looping, "watch the build") + deferred = await _new_session(client, alice) + await _run_later(client, bob, deferred, "summarize the week") + + await _fire_due_wakeups(install) + + assert {turn["cause"] for turn in install.turns} == {ALICE_PRINCIPAL} + for session_id in (looping, deferred): + processes = [p for p in cli if p.spec.session_id == session_id] + assert [_named_principals(p) for p in processes] == [[ALICE_PRINCIPAL]] + finally: + await _stop_idle_watchers(install) + + +@pytest.mark.asyncio +async def test_external_mode_a_run_now_session_names_the_person( + external_install, cli, monkeypatch, +): + """Bob's "Run now" creates the run's session for Bob, so its CLI process + names Bob on its model requests.""" + pytest.importorskip("nerve.inference_principal") + install = external_install + monkeypatch.setattr(_FakeCli, "install", install) + monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False) + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) + try: + async with _client(install.app()) as client: + response = await client.post( + "/api/cron/jobs/nightly/trigger", headers=_as(BOB_PRINCIPAL, "Bob"), + ) + assert response.status_code == 200, response.text + + (process,) = cli + assert process.spec.session_id.startswith("cron:nightly:") + assert _named_principals(process) == [BOB_PRINCIPAL] + finally: + await _stop_idle_watchers(install) diff --git a/tests/test_workflow_cron.py b/tests/test_workflow_cron.py index a288e1e45..b7adaff92 100644 --- a/tests/test_workflow_cron.py +++ b/tests/test_workflow_cron.py @@ -57,7 +57,8 @@ def _make_cron_service() -> CronService: db.get_channel_session = AsyncMock(return_value=None) db.get_session = AsyncMock(return_value=None) db.set_channel_session = AsyncMock() - # A scheduled workflow run is attributed to the agent's system actor. + # A scheduled workflow run and its run log are attributed to the + # agent's system actor. mock_system_principal(db) return CronService(config, engine, db) From 07cd1b6f4cb6cbc45c4c3ee0e16456c35c2f35b4 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:56:15 +0000 Subject: [PATCH 2/9] Show who scheduled a pending wake-up in the session sidebar The parked dot of a session that waits on a wake-up names who scheduled it: "by Alice" for a deferred message, "by Nerve for Bob" for a wakeup that the agent set in Bob's turn, "by Nerve" for autonomous work. The names come from the actor directory. A wake-up without a recorded scheduler shows only its time. The attribution hooks move from ActorLabel.tsx to hooks/useActorAttribution.ts, so that a component that needs the text for a tooltip can use the same " for " rule as ActorAttribution. Co-Authored-By: Claude Opus 5.5 --- web/src/api/websocket.ts | 2 +- web/src/components/Chat/ActorLabel.tsx | 38 +---- web/src/components/Chat/SessionSidebar.tsx | 45 +++-- .../Chat/wakeupAttribution.test.tsx | 159 ++++++++++++++++++ web/src/hooks/useActorAttribution.ts | 61 +++++++ web/src/stores/handlers/sessionHandlers.ts | 2 + web/src/types/chat.ts | 4 + 7 files changed, 260 insertions(+), 51 deletions(-) create mode 100644 web/src/components/Chat/wakeupAttribution.test.tsx create mode 100644 web/src/hooks/useActorAttribution.ts diff --git a/web/src/api/websocket.ts b/web/src/api/websocket.ts index 10b41ff5b..7f3fbba1c 100644 --- a/web/src/api/websocket.ts +++ b/web/src/api/websocket.ts @@ -39,7 +39,7 @@ export type WSMessage = // a turn can end with the session still parked on scheduled or background // work, and the sidebar redraws that row straight from the event (it skips // the list refetch for the active session). - | { type: 'session_running'; session_id: string; is_running: boolean; pending_wakeup_at?: string | null; has_background_tasks?: boolean } + | { type: 'session_running'; session_id: string; is_running: boolean; pending_wakeup_at?: string | null; pending_wakeup_created_by_actor_id?: string | null; pending_wakeup_caused_by_actor_id?: string | null; has_background_tasks?: boolean } | { type: 'session_awaiting_input'; session_id: string; awaiting: boolean } | { type: 'background_tasks_update'; session_id: string; tasks: { task_id: string; label: string; tool: string; status: 'running' | 'done' | 'failed' | 'timeout' }[] } | { type: 'workflow_progress'; session_id: string; tool_use_id: string; workflow: WorkflowSnapshot } diff --git a/web/src/components/Chat/ActorLabel.tsx b/web/src/components/Chat/ActorLabel.tsx index 61fcb7a2e..927318e41 100644 --- a/web/src/components/Chat/ActorLabel.tsx +++ b/web/src/components/Chat/ActorLabel.tsx @@ -1,38 +1,10 @@ import { - actorDiscriminators, actorName, isSystemActor, useActorStore, -} from '../../stores/actorStore'; + type Attribution, useActorAndCause, useAttribution, +} from '../../hooks/useActorAttribution'; import { useAuthStore } from '../../stores/authStore'; -import { useEffect } from 'react'; import { Badge } from '../ui'; import { Bot, User } from '../ui/icons'; -interface Attribution { - name: string; - system: boolean; - anonymous: boolean; - ambiguous: boolean; -} - -/** Resolve an actor id to its current display label. */ -function useAttribution(actorId: string | null | undefined): Attribution { - const actors = useActorStore((s) => s.actors); - const resolve = useActorStore((s) => s.resolve); - useEffect(() => { - if (actorId) resolve([actorId]); - }, [actorId, resolve]); - - const actor = actorId ? actors[actorId] : undefined; - const base = actorName(actor, actorId ?? null); - const system = isSystemActor(actor); - const discriminator = actorId ? actorDiscriminators(actors).get(actorId) : undefined; - return { - name: discriminator ? `${base} (${discriminator})` : base, - system, - anonymous: !system && !actor?.display_name?.trim() && !actor?.username?.trim(), - ambiguous: !!discriminator, - }; -} - function tooltip(verb: string, actorId: string, attribution: Attribution): string { const { name, system, anonymous, ambiguous } = attribution; if (system) return `${verb} ${name} itself — scheduled or autonomous work`; @@ -54,12 +26,8 @@ export function ActorAttribution({ actorId, causeId, verb = 'By' }: { /** Leads the tooltip, for example "Changed by". */ verb?: string; }) { - const actor = useAttribution(actorId); - const otherCauseId = actorId && causeId && causeId !== actorId ? causeId : null; - const cause = useAttribution(otherCauseId); + const { actor, forPerson, label } = useActorAndCause(actorId, causeId); if (!actorId) return null; - const forPerson = otherCauseId && !cause.system ? cause.name : null; - const label = forPerson ? `${actor.name} for ${forPerson}` : actor.name; const title = forPerson ? `${verb} ${actor.name} on behalf of ${forPerson}` : tooltip(verb, actorId, actor); diff --git a/web/src/components/Chat/SessionSidebar.tsx b/web/src/components/Chat/SessionSidebar.tsx index 4b0ac7c16..7fdf5e4b1 100644 --- a/web/src/components/Chat/SessionSidebar.tsx +++ b/web/src/components/Chat/SessionSidebar.tsx @@ -9,6 +9,7 @@ import { useChatStore } from '../../stores/chatStore'; import { useVisibleActorIds } from '../../stores/actorStore'; import { useAuthStore } from '../../stores/authStore'; import { SessionCreatorMarker } from './ActorLabel'; +import { useActorAttributionLabel } from '../../hooks/useActorAttribution'; import { useModalSurface } from '../../hooks/useModalSurface'; import { safeAreaInsets } from '../../utils/safeArea'; import { forkChat } from '../../utils/forkChat'; @@ -901,18 +902,43 @@ function reviewLoopIconTone(status: string): string { } -/** Tooltip for the parked dot: what exactly the session is waiting on. */ -function parkedTitle(session: Session): string { +/** + * Tooltip for the parked dot: what exactly the session is waiting on. + * `scheduledBy` names who scheduled the wake-up ("Alice", "Nerve for Bob"), + * or is null when that was not recorded. + */ +function parkedTitle(session: Session, scheduledBy: string | null): string { const parts: string[] = []; if (session.has_background_tasks) parts.push('Background job running'); if (session.pending_wakeup_at) { const at = parseTimestamp(session.pending_wakeup_at); - parts.push(`Scheduled wake-up at ${at.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' })}`); + const time = at.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); + parts.push(`Scheduled wake-up at ${time}${scheduledBy ? ` by ${scheduledBy}` : ''}`); } return parts.join(' · ') || 'Waiting on pending work'; } +/** + * Parked: no turn in flight, but a wake-up is scheduled or a background job + * is still running. Same pulsing dot as running (the session isn't done), + * in violet — the one hue no other indicator uses — so "working right now" + * stays distinguishable from "will pick itself back up". + */ +function ParkedIndicator({ session }: { session: Session }) { + const scheduledBy = useActorAttributionLabel( + session.pending_wakeup_at ? session.pending_wakeup_created_by_actor_id : null, + session.pending_wakeup_caused_by_actor_id, + ); + return ( + + + + + ); +} + + /** Pulsing dot for running sessions, solid dot for other notable states. */ function StatusIndicator({ session, isActive, isRunning }: { session: Session; @@ -976,18 +1002,7 @@ function StatusIndicator({ session, isActive, isRunning }: { return ; } - // Parked: no turn in flight, but a wake-up is scheduled or a background job - // is still running. Same pulsing dot as running (the session isn't done), - // in violet — the one hue no other indicator uses — so "working right now" - // stays distinguishable from "will pick itself back up". - if (isParked(session)) { - return ( - - - - - ); - } + if (isParked(session)) return ; // Stopped: solid yellow if (session.status === 'stopped') { diff --git a/web/src/components/Chat/wakeupAttribution.test.tsx b/web/src/components/Chat/wakeupAttribution.test.tsx new file mode 100644 index 000000000..d97e966b2 --- /dev/null +++ b/web/src/components/Chat/wakeupAttribution.test.tsx @@ -0,0 +1,159 @@ +// @vitest-environment jsdom +import { render, waitFor } from '@testing-library/react'; +import { MemoryRouter } from 'react-router-dom'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ActorRef } from '../../api/client'; +import type { WSMessage } from '../../api/websocket'; +import type { Session } from '../../types/chat'; + +function installStorage(): void { + const data = new Map(); + const storage = { + getItem: (k: string) => data.get(k) ?? null, + setItem: (k: string, v: string) => void data.set(k, String(v)), + removeItem: (k: string) => void data.delete(k), + clear: () => data.clear(), + key: (i: number) => [...data.keys()][i] ?? null, + get length() { return data.size; }, + }; + for (const target of [globalThis, globalThis.window]) { + if (target) Object.defineProperty(target, 'localStorage', { + value: storage, configurable: true, writable: true, + }); + } +} +installStorage(); + +vi.mock('../../api/client', () => ({ + api: { + listActors: vi.fn(), listSessions: vi.fn(), listArchivedSessions: vi.fn(), + listSystemSessions: vi.fn(), searchSessions: vi.fn(), + }, + getToken: vi.fn(() => 'tok'), setToken: vi.fn(), clearToken: vi.fn(), + setUnauthorizedHandler: vi.fn(), +})); +vi.mock('../../api/websocket', () => ({ + ws: { + switchSession: vi.fn(), send: vi.fn(), connect: vi.fn(), + disconnect: vi.fn(), onMessage: vi.fn(() => () => {}), + }, +})); + +const { api } = await import('../../api/client'); +const { useActorStore } = await import('../../stores/actorStore'); +const { useAuthStore } = await import('../../stores/authStore'); +const { useChatStore } = await import('../../stores/chatStore'); +const { handleSessionRunning } = await import('../../stores/handlers/sessionHandlers'); +const { SessionSidebar } = await import('./SessionSidebar'); + +const listActors = api.listActors as unknown as ReturnType; +const ALICE = 'actor-alice'; +const BOB = 'actor-bob'; +const SYSTEM = 'actor-system'; + +function actorRef(id: string, overrides: Partial = {}): ActorRef { + return { id, kind: 'human', display_name: null, username: null, ...overrides } as ActorRef; +} + +function parked( + id: string, + createdBy: string | null, + causedBy: string | null, +): Session { + return { + id, title: id, source: 'web', updated_at: new Date().toISOString(), + starred: true, is_running: false, + pending_wakeup_at: new Date(Date.now() + 30 * 60_000).toISOString(), + pending_wakeup_created_by_actor_id: createdBy, + pending_wakeup_caused_by_actor_id: causedBy, + }; +} + +function renderSidebar(sessions: Session[]) { + return render( + + {}} onDelete={() => {}} + /> + , + ); +} + +/** The parked dots' tooltips, without the local fire time. */ +function wakeupTooltips(): string[] { + return [...document.querySelectorAll('[title^="Scheduled wake-up at"]')] + .map((el) => (el.getAttribute('title') ?? '').replace(/ at [^ ]+( [AP]M)?/, ' at T')); +} + +beforeEach(() => { + vi.clearAllMocks(); + useActorStore.getState().reset(); + useAuthStore.setState({ + viewer: actorRef(BOB), + account: { id: 'account-bob', username: null }, + }); + useChatStore.setState({ + sessions: [], searchResults: null, archivedSessions: null, systemSessions: null, + archivedCount: 0, systemCount: 0, drafts: {}, reads: {}, readsBaseline: 0, + virtualSession: null, activeSession: '', + }); + (api.listSessions as unknown as ReturnType).mockResolvedValue({ + sessions: [], archived_count: 0, system_count: 0, has_more: false, next_offset: 0, + }); + listActors.mockResolvedValue({ + actors: [ + actorRef(ALICE, { display_name: 'Alice' }), + actorRef(BOB, { display_name: 'Bob' }), + actorRef(SYSTEM, { kind: 'system' }), + ], + }); +}); + +describe('who scheduled a pending wake-up', () => { + it('names the person, the agent for a person, or the agent alone', async () => { + renderSidebar([ + parked('deferred', ALICE, ALICE), + parked('agent-for-bob', SYSTEM, BOB), + parked('cron-loop', SYSTEM, SYSTEM), + ]); + + await waitFor(() => expect(wakeupTooltips()).toEqual([ + 'Scheduled wake-up at T by Alice', + 'Scheduled wake-up at T by Nerve for Bob', + 'Scheduled wake-up at T by Nerve', + ])); + }); + + it('names nobody for a wake-up whose scheduler was not recorded', async () => { + renderSidebar([parked('legacy', null, null)]); + + await waitFor(() => expect(wakeupTooltips()).toEqual(['Scheduled wake-up at T'])); + expect(listActors).not.toHaveBeenCalled(); + }); + + it('takes the scheduler from the session_running event', () => { + useChatStore.setState({ sessions: [parked('s1', ALICE, ALICE)] }); + const event = { + type: 'session_running', session_id: 's1', is_running: false, + pending_wakeup_at: '2026-01-01T00:10:00Z', + pending_wakeup_created_by_actor_id: SYSTEM, + pending_wakeup_caused_by_actor_id: BOB, + } as Extract; + + handleSessionRunning(event, useChatStore.getState, useChatStore.setState); + + const [s1] = useChatStore.getState().sessions; + expect(s1.pending_wakeup_created_by_actor_id).toBe(SYSTEM); + expect(s1.pending_wakeup_caused_by_actor_id).toBe(BOB); + + handleSessionRunning( + { type: 'session_running', session_id: 's1', is_running: false }, + useChatStore.getState, useChatStore.setState, + ); + const [cleared] = useChatStore.getState().sessions; + expect(cleared.pending_wakeup_at).toBeNull(); + expect(cleared.pending_wakeup_created_by_actor_id).toBeNull(); + expect(cleared.pending_wakeup_caused_by_actor_id).toBeNull(); + }); +}); diff --git a/web/src/hooks/useActorAttribution.ts b/web/src/hooks/useActorAttribution.ts new file mode 100644 index 000000000..0458cda61 --- /dev/null +++ b/web/src/hooks/useActorAttribution.ts @@ -0,0 +1,61 @@ +import { useEffect } from 'react'; +import { + actorDiscriminators, actorName, isSystemActor, useActorStore, +} from '../stores/actorStore'; + +export interface Attribution { + name: string; + system: boolean; + anonymous: boolean; + ambiguous: boolean; +} + +/** Resolve an actor id to its current display label. */ +export function useAttribution(actorId: string | null | undefined): Attribution { + const actors = useActorStore((s) => s.actors); + const resolve = useActorStore((s) => s.resolve); + useEffect(() => { + if (actorId) resolve([actorId]); + }, [actorId, resolve]); + + const actor = actorId ? actors[actorId] : undefined; + const base = actorName(actor, actorId ?? null); + const system = isSystemActor(actor); + const discriminator = actorId ? actorDiscriminators(actors).get(actorId) : undefined; + return { + name: discriminator ? `${base} (${discriminator})` : base, + system, + anonymous: !system && !actor?.display_name?.trim() && !actor?.username?.trim(), + ambiguous: !!discriminator, + }; +} + +/** + * The actor, and the name of the person the actor worked for when that is + * another person: the system actor (Nerve) doing work that a person started. + * A cause that is the actor itself, or the system actor, adds nothing. + * `label` is "Alice" or "Nerve for Alice". + */ +export function useActorAndCause( + actorId: string | null | undefined, + causeId: string | null | undefined, +): { actor: Attribution; forPerson: string | null; label: string } { + const actor = useAttribution(actorId); + const otherCauseId = actorId && causeId && causeId !== actorId ? causeId : null; + const cause = useAttribution(otherCauseId); + const forPerson = otherCauseId && !cause.system ? cause.name : null; + const label = forPerson ? `${actor.name} for ${forPerson}` : actor.name; + return { actor, forPerson, label }; +} + +/** + * The text of `ActorAttribution` for a place that needs a string, such as a + * tooltip. `null` without an actor id. + */ +export function useActorAttributionLabel( + actorId: string | null | undefined, + causeId: string | null | undefined, +): string | null { + const { label } = useActorAndCause(actorId, causeId); + return actorId ? label : null; +} diff --git a/web/src/stores/handlers/sessionHandlers.ts b/web/src/stores/handlers/sessionHandlers.ts index 344206b20..2b592fbb0 100644 --- a/web/src/stores/handlers/sessionHandlers.ts +++ b/web/src/stores/handlers/sessionHandlers.ts @@ -230,6 +230,8 @@ export function handleSessionRunning( ...sess, is_running: msg.is_running, pending_wakeup_at: msg.pending_wakeup_at ?? null, + pending_wakeup_created_by_actor_id: msg.pending_wakeup_created_by_actor_id ?? null, + pending_wakeup_caused_by_actor_id: msg.pending_wakeup_caused_by_actor_id ?? null, has_background_tasks: !!msg.has_background_tasks, }); set(s => { diff --git a/web/src/types/chat.ts b/web/src/types/chat.ts index af824d119..3fa46102e 100644 --- a/web/src/types/chat.ts +++ b/web/src/types/chat.ts @@ -140,6 +140,10 @@ export interface Session { // most one per session); `has_background_tasks` covers run_in_background // Bash/Agent work still live inside the session's CLI. pending_wakeup_at?: string | null; + // Who scheduled the pending wake-up (a person, or the system actor when + // the agent did) and on whose behalf it runs. Null when not recorded. + pending_wakeup_created_by_actor_id?: string | null; + pending_wakeup_caused_by_actor_id?: string | null; has_background_tasks?: boolean; starred?: boolean; // Live review-loop state for this observer session (rehydrated from From a19b9db0ab66dfc643dd51670f7e067180ee1c17 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:56:56 +0000 Subject: [PATCH 3/9] Document who schedules wakeups and starts cron runs docs/accounts.md lists the four new attribution columns and the actor and cause of "Run now", "Run later", agent wakeups and the turns they fire. docs/api.md describes the actor fields of cron logs, the trigger route and the pending wakeup fields of session rows. docs/cron.md lists the two cron_logs columns. Co-Authored-By: Claude Opus 5.5 --- docs/accounts.md | 13 ++++++++++++- docs/api.md | 26 +++++++++++++++++++++++++- docs/cron.md | 7 +++++++ 3 files changed, 44 insertions(+), 2 deletions(-) diff --git a/docs/accounts.md b/docs/accounts.md index ee16e3f6c..1e780381e 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -203,6 +203,8 @@ These columns store actor IDs: | `review_loops.created_by_actor_id`, `review_loops.caused_by_actor_id` | Actor and cause of a review loop. Each leg session of the loop is created with this cause | | `review_loops.killed_by_actor_id`, `review_loops.killed_caused_by_actor_id` | Actor and cause of the kill of a review loop | | `review_loop_decisions.actor_id`, `review_loop_decisions.caused_by_actor_id` | Actor and cause of one decision on a parked review loop | +| `session_wakeups.created_by_actor_id`, `session_wakeups.caused_by_actor_id` | Who scheduled a wakeup or a deferred message, and on whose behalf. The turn that it fires runs in its session, for the session's cause | +| `cron_logs.actor_id`, `cron_logs.caused_by_actor_id` | Who started a cron or source run, and on whose behalf | `nerve/db/actor_columns.py` lists every attribution column. Clients resolve names through `GET /api/actors`, which returns every actor that these columns @@ -218,6 +220,7 @@ never changes it: |---|---| | A person creates or forks a session (web UI, REST API, browser WebSocket) | That person | | A cron job or a hook session | The system actor | +| A cron job session that a person's "Run now" creates (an isolated run, or a new generation of a persistent job) | That person. A persistent job's current generation keeps its cause | | A plan implementation session after an approval through the REST API | The person who approved | | A plan implementation session after the agent approved the plan | The cause of the session in which the agent approved it | | A planner session that a revision request creates | The person who asked for the revision | @@ -237,7 +240,11 @@ This is an approximation in two places: attributed to the person who started the session, also in turns that another person sent. - A deferred message ("run later") or a wakeup that runs in an existing session - is attributed to that session's cause. + is attributed to that session's cause, also when another person deferred + the message. +- A persistent cron job reuses its current session. When a person's "Run + now" started that session, the later scheduled runs in it are attributed + to that person until the session rotates. In external mode, the model requests of a session name the session's cause in the same way, so the same approximation applies to model usage. See @@ -254,6 +261,10 @@ in the same way, so the same approximation applies to model usage. See | A person starts or kills a workflow run or a review loop through the web UI or the REST API, or decides on a parked loop through the REST API or its decision card in the web UI | The person | The person | | The agent starts or kills a workflow run or a review loop | The system actor | The session's cause | | The message that relays a web answer into the session that asked | The person who answered | Not recorded (the agent's work in that turn has the session's cause) | +| A scheduled cron or source run (also a catch-up or missed run, and `nerve cron `) | The system actor on the run log | The system actor | +| A person starts a cron or source run from the web UI or with `POST /api/cron/jobs/{job_id}/trigger` | The person on the run log | The person | +| A person defers a message with "Run later" | The person on the wakeup row | The person | +| The agent schedules a wakeup (`ScheduleWakeup`, `schedule_wakeup`) | The system actor | The session's cause | | Telegram, Slack, or imported Codex input, and channel commands (`/new`, `/stop`, `/star`) | None until a local identity mapping exists | None | | An answer through Telegram, Slack, or a hosted channel (also a decision on a review-loop card), and the message that relays it | None | None | | Assistant and tool output | None | None | diff --git a/docs/api.md b/docs/api.md index 86892befa..d4b80e101 100644 --- a/docs/api.md +++ b/docs/api.md @@ -795,6 +795,13 @@ whose behalf all agent work in the session runs. Nerve sets it when it creates the session and never changes it (an actor id, or `null` when unknown). See [Attribution](accounts.md#attribution). +Session list rows carry `pending_wakeup_at`, the fire time of the session's +pending wakeup or deferred message (`null` when there is none), with +`pending_wakeup_created_by_actor_id` (who scheduled it: a person, or the +system actor when the agent did) and `pending_wakeup_caused_by_actor_id` (on +whose behalf it was scheduled; the turn that it fires runs for the session's +cause). The `session_running` WebSocket event carries the same three fields. + #### `GET /api/sessions/{id}` Get session details. @@ -1321,15 +1328,32 @@ combine with `offset` for pagination. Each log row carries the `session_id` of the chat session the run executed in (null for source runners and missed runs). +Each log row also carries `actor_id` (who started the run) and +`caused_by_actor_id` (on whose behalf). Both are the system actor for a +scheduled run, and the person for a run that a person started with +`POST /api/cron/jobs/{job_id}/trigger`. Both are `null` on older rows. See +[Attribution](accounts.md#attribution). + ```json Response: { - "logs": [ { "id": 12, "job_id": "...", "status": "success", "session_id": "cron:...", ... } ], + "logs": [ { "id": 12, "job_id": "...", "status": "success", "session_id": "cron:...", "actor_id": "…", "caused_by_actor_id": "…", ... } ], "total": 234, "limit": 50, "offset": 0 } ``` +#### `POST /api/cron/jobs/{job_id}/trigger` +Run a cron job or a source runner now, and wait until the run ends. The run +log records the signed-in person as actor and cause. A session that the run +creates (an isolated run, or a new generation of a persistent job) has the +person as its cause; a persistent job's current session keeps its cause. +Returns 404 for an unknown job. + +```json +Response: { "job_id": "morning-briefing", "triggered": true } +``` + ### Health #### `GET /health` diff --git a/docs/cron.md b/docs/cron.md index e00ad2975..9940608be 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -512,6 +512,13 @@ Every cron and source run is logged in the `cron_logs` SQLite table: - `status` — `success`, `error`, or `missed` (see [Late runs](#late-runs)) - `output` — First 2000 chars of response / summary - `error` — Error message if failed +- `actor_id` / `caused_by_actor_id` — Who started the run, and on whose + behalf: the system actor for a scheduled, catch-up or missed run, or for + `nerve cron `; the person who triggered the run through the web UI or + `POST /api/cron/jobs/{job_id}/trigger`. A session that a person's run + creates has that person as its cause, so the agent's work in it is the + system actor's for that person. See + [Attribution](accounts.md#attribution). Source runs also log to `source_run_log` with per-source diagnostics (records fetched/processed, errors). From 1c4975c6453de398ecd1c9f82768a8bf7d95c74a Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:30:41 +0000 Subject: [PATCH 4/9] Use no em-dashes in the schedule attribution text The writing rules forbid em-dashes in docs and comments. Replace the four in the text that this branch adds with a colon, parentheses or a comma. Co-Authored-By: Claude Opus 5.5 --- docs/cron.md | 2 +- nerve/agent/engine.py | 4 ++-- tests/test_schedule_attribution.py | 2 +- web/src/components/Chat/SessionSidebar.tsx | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/cron.md b/docs/cron.md index 9940608be..71fd6594d 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -512,7 +512,7 @@ Every cron and source run is logged in the `cron_logs` SQLite table: - `status` — `success`, `error`, or `missed` (see [Late runs](#late-runs)) - `output` — First 2000 chars of response / summary - `error` — Error message if failed -- `actor_id` / `caused_by_actor_id` — Who started the run, and on whose +- `actor_id` / `caused_by_actor_id`: who started the run, and on whose behalf: the system actor for a scheduled, catch-up or missed run, or for `nerve cron `; the person who triggered the run through the web UI or `POST /api/cron/jobs/{job_id}/trigger`. A session that a person's run diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index d34913230..6ae3cbf3f 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -1584,8 +1584,8 @@ async def pending_wakeup(self, session_id: str) -> dict | None: ``ScheduleWakeup`` keeps at most one pending row per session (a new request supersedes the old one), so this is a single row rather - than a list. Never raises — a listing/indicator caller must not fail - on an enrichment read. + than a list. Never raises: a listing or indicator caller must not + fail on an enrichment read. """ try: rows = await self.db.list_pending_wakeups(session_id) diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py index c42aaa25f..783a152a7 100644 --- a/tests/test_schedule_attribution.py +++ b/tests/test_schedule_attribution.py @@ -29,7 +29,7 @@ from nerve.cron.service import CronService from nerve.db.actor_columns import ACTOR_COLUMNS from nerve.identity import Attribution -from tests.test_cause_attribution import ( # noqa: F401 — fixtures +from tests.test_cause_attribution import ( # noqa: F401 (fixtures) ALICE_PRINCIPAL, BOB_PRINCIPAL, _as, diff --git a/web/src/components/Chat/SessionSidebar.tsx b/web/src/components/Chat/SessionSidebar.tsx index 7fdf5e4b1..fa59b5256 100644 --- a/web/src/components/Chat/SessionSidebar.tsx +++ b/web/src/components/Chat/SessionSidebar.tsx @@ -922,7 +922,7 @@ function parkedTitle(session: Session, scheduledBy: string | null): string { /** * Parked: no turn in flight, but a wake-up is scheduled or a background job * is still running. Same pulsing dot as running (the session isn't done), - * in violet — the one hue no other indicator uses — so "working right now" + * in violet (the one hue that no other indicator uses), so "working right now" * stays distinguishable from "will pick itself back up". */ function ParkedIndicator({ session }: { session: Session }) { From a60164fee4de7d01eb837f7d1ce71aaf67935c10 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:57:13 +0000 Subject: [PATCH 5/9] Run the inference principal tests of wakeups without a skip The inference principal header is in the base of this branch, so the tests of the header on fired wakeups and on a run-now session run unconditionally. Co-Authored-By: Claude Opus 5.5 --- tests/test_schedule_attribution.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py index 783a152a7..248de6a47 100644 --- a/tests/test_schedule_attribution.py +++ b/tests/test_schedule_attribution.py @@ -29,6 +29,7 @@ from nerve.cron.service import CronService from nerve.db.actor_columns import ACTOR_COLUMNS from nerve.identity import Attribution +from nerve.inference_principal import CUSTOM_HEADERS_ENV, INFERENCE_PRINCIPAL_HEADER from tests.test_cause_attribution import ( # noqa: F401 (fixtures) ALICE_PRINCIPAL, BOB_PRINCIPAL, @@ -633,8 +634,6 @@ async def test_external_mode_run_later_and_run_now_name_the_principal( def _named_principals(process: _FakeCli) -> list[str]: """The principal headers that the CLI process sends on model requests.""" - from nerve.inference_principal import CUSTOM_HEADERS_ENV, INFERENCE_PRINCIPAL_HEADER - named = [] for line in process.options.env.get(CUSTOM_HEADERS_ENV, "").splitlines(): name, _, value = line.partition(":") @@ -653,10 +652,9 @@ async def test_external_mode_wakeup_turns_name_the_session_cause_to_the_model_ap message in another of Alice's sessions. Both fire in Alice's sessions, whose CLI processes name Alice, the session cause. """ - pytest.importorskip("nerve.inference_principal") install = external_install monkeypatch.setattr(_FakeCli, "install", install) - monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False) + monkeypatch.delenv(CUSTOM_HEADERS_ENV, raising=False) alice, bob = _as(ALICE_PRINCIPAL, "Alice"), _as(BOB_PRINCIPAL, "Bob") try: async with _client(install.app()) as client: @@ -683,10 +681,9 @@ async def test_external_mode_a_run_now_session_names_the_person( ): """Bob's "Run now" creates the run's session for Bob, so its CLI process names Bob on its model requests.""" - pytest.importorskip("nerve.inference_principal") install = external_install monkeypatch.setattr(_FakeCli, "install", install) - monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False) + monkeypatch.delenv(CUSTOM_HEADERS_ENV, raising=False) job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) try: From 9cadd7a92a4f29b4393dbbf77d354079083bdcb9 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:00:44 +0000 Subject: [PATCH 6/9] Start a workflow-run job's run for the person who used "Run now" A cron job that declares a workflow run started the run for the system actor also when a person triggered the job through the REST API, so the run's turns and model requests named the system principal. The cron service now passes the cause of the cron run to the run: the system actor starts it, for the person who used "Run now", or for itself on a scheduled run. docs/accounts.md, docs/api.md and docs/workflow-runs.md say so, and the actor directory intro in docs/api.md names wakeups and cron run logs. Co-Authored-By: Claude Opus 5.5 --- docs/accounts.md | 2 +- docs/api.md | 9 ++++--- docs/workflow-runs.md | 6 +++-- nerve/cron/service.py | 18 ++++++++++--- tests/test_schedule_attribution.py | 43 ++++++++++++++++++++++++++++++ tests/test_workflow_cron.py | 14 ++++++++++ 6 files changed, 81 insertions(+), 11 deletions(-) diff --git a/docs/accounts.md b/docs/accounts.md index 1e780381e..996016410 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -262,7 +262,7 @@ in the same way, so the same approximation applies to model usage. See | The agent starts or kills a workflow run or a review loop | The system actor | The session's cause | | The message that relays a web answer into the session that asked | The person who answered | Not recorded (the agent's work in that turn has the session's cause) | | A scheduled cron or source run (also a catch-up or missed run, and `nerve cron `) | The system actor on the run log | The system actor | -| A person starts a cron or source run from the web UI or with `POST /api/cron/jobs/{job_id}/trigger` | The person on the run log | The person | +| A person starts a cron or source run from the web UI or with `POST /api/cron/jobs/{job_id}/trigger` | The person on the run log; the system actor for the workflow run that a workflow-run job starts | The person | | A person defers a message with "Run later" | The person on the wakeup row | The person | | The agent schedules a wakeup (`ScheduleWakeup`, `schedule_wakeup`) | The system actor | The session's cause | | Telegram, Slack, or imported Codex input, and channel commands (`/new`, `/stop`, `/star`) | None until a local identity mapping exists | None | diff --git a/docs/api.md b/docs/api.md index d4b80e101..7f1836dc5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -714,8 +714,8 @@ writes no configuration. An actor is *who* something is attributed to: a person, or the agent's system principal. Sessions, messages, task events, notifications, silences, plans, -workflow runs, review loops and review-loop decisions store an actor id, -never a name, so this is where a name is looked up at +workflow runs, review loops, review-loop decisions, wakeups and cron run logs +store an actor id, never a name, so this is where a name is looked up at render time. A rename changes every label and moves no stored row. @@ -1347,8 +1347,9 @@ Response: { Run a cron job or a source runner now, and wait until the run ends. The run log records the signed-in person as actor and cause. A session that the run creates (an isolated run, or a new generation of a persistent job) has the -person as its cause; a persistent job's current session keeps its cause. -Returns 404 for an unknown job. +person as its cause; a persistent job's current session keeps its cause. The +workflow run that a workflow-run job starts is the system actor's, for the +person. Returns 404 for an unknown job. ```json Response: { "job_id": "morning-briefing", "triggered": true } diff --git a/docs/workflow-runs.md b/docs/workflow-runs.md index 319e63703..d21c3b8a5 100644 --- a/docs/workflow-runs.md +++ b/docs/workflow-runs.md @@ -102,8 +102,10 @@ A run records who created it and on whose behalf - Through `workflow_run_start` or `workflow_run_kill`, the actor is the system actor and the cause is the cause of the session that called the tool. -- A cron job's run is the system actor's, for itself. A review-loop leg is - the system actor's, for the cause of the loop. +- A cron job's run is the system actor's, for itself. When a person starts + the job with "Run now" (`POST /api/cron/jobs/{job_id}/trigger`), the run + is the system actor's, for that person. A review-loop leg is the system + actor's, for the cause of the loop. - The run's session is created with the run's cause, so everything the agent does in it (and, in external mode, its model requests) is for that cause. A run with no cause (one started by input that names no person) diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 419cd6331..882b588ba 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -1170,7 +1170,7 @@ async def _run_job_inner( # Workflow-run job: launch a budget-capped workflow run and # return — the run executes and notifies on its own, so none # of the cron session machinery below applies. - await self._start_workflow_run(job, log_id) + await self._start_workflow_run(job, log_id, attribution) return session_id: str | None = None @@ -1266,16 +1266,25 @@ async def _run_job_inner( log_id, "error", error=str(e), session_id=session_id, ) - async def _start_workflow_run(self, job: CronJob, log_id: int) -> None: + async def _start_workflow_run( + self, job: CronJob, log_id: int, attribution: Attribution | None = None, + ) -> None: """Start the workflow run a job declares instead of a prompt. Fire-and-forget: the cron log records only the launch. The run owns its ``workflow:`` session and sends its own budget / completion / failure notifications — no cron session is created and the job does not wait for the run to finish. + + ``attribution`` is the run log's (see :meth:`_run_job_wrapper`). + The cron service starts the run, so its actor is the system actor; + its cause is the cause of the cron run, and the run's turns run for + it. """ from nerve.workflows import get_workflow_run_service + cause_id = (attribution or self._own_attribution()).caused_by_actor_id + service = get_workflow_run_service() if service is None: logger.warning( @@ -1322,8 +1331,9 @@ async def _start_workflow_run(self, job: CronJob, log_id: int) -> None: budget_usd=w["budget_usd"], title=w.get("title") or job.id, created_by=f"cron:{job.id}", - # A scheduled job is autonomous work: system for system. - attribution=attribution_for(self.db.system_actor), + # A scheduled job is autonomous work: system for system. A + # run that a person started is the system's, for that person. + attribution=attribution_for(self.db.system_actor, cause_id), ) session_id = run.get("session_id") or ("workflow:" + run["id"]) await self.db.set_cron_log_session(log_id, session_id) diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py index 248de6a47..c0562c78d 100644 --- a/tests/test_schedule_attribution.py +++ b/tests/test_schedule_attribution.py @@ -453,6 +453,49 @@ async def test_run_now_through_rest_runs_as_the_system_for_the_person( assert await install.session_cause_of(turn["session_id"]) == alice +@pytest.mark.asyncio +async def test_run_now_of_a_workflow_job_starts_the_run_for_the_person( + install, monkeypatch, +): + """A job that declares a workflow run starts it for the person who used + "Run now"; a scheduled run of the same job starts it for the system.""" + import nerve.workflows + + alice = install.actor_id("alice") + system = install.system_actor_id + started: list[dict] = [] + + async def _start_run(**kwargs): + started.append(kwargs) + return {"id": f"wfr-{len(started)}", "budget_usd": 1.0} + + monkeypatch.setattr( + nerve.workflows, "_service", SimpleNamespace(start_run=_start_run, db=install.db), + ) + job = CronJob( + id="audit", schedule="1h", + workflow={"engine": "claude-workflow", "prompt": "audit", "budget_usd": 1.0}, + ) + service = _cron_service(install, job) + monkeypatch.setattr("nerve.gateway.server._cron_service", service) + + async with _client(install.app()) as client: + response = await client.post( + "/api/cron/jobs/audit/trigger", headers=install.headers("alice"), + ) + assert response.status_code == 200, response.text + await service._run_job_wrapper(job) + + assert [run["attribution"] for run in started] == [ + Attribution(system, alice), + Attribution(system, system), + ] + assert await _cron_logs(install) == [ + ("audit", "success", alice, alice), + ("audit", "success", system, system), + ] + + @pytest.mark.asyncio async def test_run_now_of_a_source_records_the_person(install, monkeypatch): bob = install.actor_id("bob") diff --git a/tests/test_workflow_cron.py b/tests/test_workflow_cron.py index b7adaff92..aee1a9593 100644 --- a/tests/test_workflow_cron.py +++ b/tests/test_workflow_cron.py @@ -212,6 +212,20 @@ async def test_starts_run_with_declared_spec( ), ) + async def test_a_run_that_a_person_started_runs_for_that_person( + self, cron_service, workflow_service, + ): + alice = "00000000-0000-4000-8000-0000000a11ce" + + await cron_service._run_job_inner( + _workflow_job(), Attribution(actor_id=alice, caused_by_actor_id=alice), + ) + + # The cron service starts the run on Alice's behalf. + assert workflow_service.start_run.await_args.kwargs["attribution"] == ( + Attribution(actor_id=FAKE_SYSTEM_ACTOR.actor_id, caused_by_actor_id=alice) + ) + async def test_links_run_session_and_logs_success( self, cron_service, workflow_service, ): From 45e39c384a6c3f39b262092a58ce99a545d618d7 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Sat, 17 Oct 2026 11:01:53 +0000 Subject: [PATCH 7/9] Run a fired wakeup for its session's cause Under the session cause model a session has one cause, and every agent turn in it runs for that cause. A fired wakeup runs in its session, so the cron service no longer passes the wakeup's stored cause to the turn; the wakeup row keeps who scheduled it and for whom. A deferred message that Bob leaves in Alice's session therefore fires for Alice, the documented approximation. The comments of the wakeup store, the migration, the ScheduleWakeup hook and the schedule_wakeup tool say that an agent wakeup records the session's cause. Tests: the wakeup sweep passes no cause to the turn. The cron rotation test seeds a person as the session cause and expects the retitle to be the system actor's for that person. A persistent session that a person's "Run now" creates keeps that person as cause for a later scheduled run, its rotation retitle names the person, and the next session is the system actor's. Co-Authored-By: Claude Opus 5.5 --- nerve/agent/engine.py | 9 ++- nerve/agent/tools/handlers/wakeups.py | 2 +- nerve/cron/service.py | 8 +-- .../migrations/v057_schedule_attribution.py | 12 ++-- nerve/db/wakeups.py | 7 +-- tests/test_cron.py | 15 +++-- tests/test_schedule_attribution.py | 62 +++++++++++++++++++ tests/test_wakeups.py | 3 + 8 files changed, 92 insertions(+), 26 deletions(-) diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 6ae3cbf3f..335c482a2 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -1068,9 +1068,8 @@ async def _record_wakeup( ) -> int | None: """Persist a ScheduleWakeup request from its tool input. - ``attribution`` is that of the tool call: the system actor, and the - cause of the turn that made the call. The wakeup turn runs for that - cause. + ``attribution`` is that of the tool call: the system actor and the + session's cause. Returns the new wakeup id, or ``None`` when there's no prompt to re-inject (in which case nothing is scheduled). @@ -1358,8 +1357,8 @@ async def _get_or_create_client( ) async def _record_wakeup_cb(sid: str, tool_input: dict) -> Any: - # The CLI runs the hook inside the turn that called the tool, - # so the session's turn cause is the cause of that call. + # An agent action in a session: the system actor, for the + # session's cause. return await self._record_wakeup( self.db, sid, tool_input, attribution=attribution_for( diff --git a/nerve/agent/tools/handlers/wakeups.py b/nerve/agent/tools/handlers/wakeups.py index eba4a5350..53c9e072c 100644 --- a/nerve/agent/tools/handlers/wakeups.py +++ b/nerve/agent/tools/handlers/wakeups.py @@ -81,7 +81,7 @@ async def schedule_wakeup_handler(ctx: ToolContext, args: dict) -> ToolResult: is_error=True, ) - # The wakeup turn runs for the cause of the turn that schedules it. + # An agent action in a session: the system actor, for the session's cause. wakeup_id = await ctx.engine._record_wakeup( ctx.db, ctx.session_id, args, attribution=ctx.attribution, ) diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 882b588ba..5be9bc9ce 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -1455,10 +1455,9 @@ def _dispatch_wakeup(self, session_id: str, wakeup: dict) -> None: # The actor is the instance's own: a model-scheduled wakeup, or a # deferral this service is delivering. Who wrote a run-later message # is already recorded on the row the route persisted when they - # composed it. The cause is the one stored when the wakeup was - # scheduled: the person who deferred the message, or the cause of - # the turn that called ScheduleWakeup. A stored NULL (the cause was - # unknown, or the row is older than v057) stays unknown. + # composed it. The turn runs in the wakeup's session, so its cause + # is the session's cause; the wakeup row keeps who scheduled it and + # for whom. task = asyncio.create_task( self.engine.run( session_id=session_id, @@ -1466,7 +1465,6 @@ def _dispatch_wakeup(self, session_id: str, wakeup: dict) -> None: source=source, internal=True, actor=self.db.system_actor, - cause=wakeup.get("caused_by_actor_id"), # A run-later deferral is a message the user wrote and asked # to be delivered now, so it belongs back at the top of the # sidebar. A model's own ScheduleWakeup tick is the session diff --git a/nerve/db/migrations/v057_schedule_attribution.py b/nerve/db/migrations/v057_schedule_attribution.py index 01d709e6b..b27e0cbff 100644 --- a/nerve/db/migrations/v057_schedule_attribution.py +++ b/nerve/db/migrations/v057_schedule_attribution.py @@ -4,11 +4,10 @@ - ``session_wakeups.created_by_actor_id``: who scheduled the wakeup. A person who deferred a message with "Run later", or the system actor when the agent - called ``ScheduleWakeup`` / ``schedule_wakeup`` in a turn. -- ``session_wakeups.caused_by_actor_id``: on whose behalf the wakeup turn - runs. The person for "Run later", the cause of the turn that called the - tool for an agent wakeup. The turn that the wakeup fires runs for this - cause. + called ``ScheduleWakeup`` / ``schedule_wakeup`` in a session. +- ``session_wakeups.caused_by_actor_id``: on whose behalf it was scheduled. + The person for "Run later", the session's cause for an agent wakeup. The + turn that the wakeup fires runs in its session, for the session's cause. - ``cron_logs.actor_id``: who started the run. The system actor for a scheduled run, a catch-up run or a missed run; the person who used "Run now" through the REST API. @@ -16,8 +15,7 @@ system actor for scheduled work, the person for "Run now". **Existing rows stay NULL and nothing is backfilled.** NULL reads as "not -recorded". A wakeup row from before this migration fires with an unknown -cause. +recorded". The foreign keys follow v048: an ID that resolves to no actor can never be stored, and NULL is exempt from the check. Each column is added only when the diff --git a/nerve/db/wakeups.py b/nerve/db/wakeups.py index e90c74c48..e120ab3ad 100644 --- a/nerve/db/wakeups.py +++ b/nerve/db/wakeups.py @@ -6,8 +6,8 @@ through ``engine.run(..., source="wakeup")`` at ``fire_at``. Each row records who scheduled it (``created_by_actor_id``) and on whose -behalf it runs (``caused_by_actor_id``). The turn that the wakeup fires runs -for that cause. +behalf (``caused_by_actor_id``). The turn that the wakeup fires runs in its +session, so it has the session's cause. """ from __future__ import annotations @@ -33,8 +33,7 @@ async def add_wakeup( ``/loop`` self-pacing model re-calls the tool each turn). So a new request supersedes any earlier pending wakeup for the session. - ``attribution`` names who scheduled the wakeup and the cause that - its turn runs for. + ``attribution`` names who scheduled the wakeup and on whose behalf. Returns the new wakeup id. """ diff --git a/tests/test_cron.py b/tests/test_cron.py index ba88803c3..2239d49f9 100644 --- a/tests/test_cron.py +++ b/tests/test_cron.py @@ -19,9 +19,12 @@ _parse_interval, _parse_timestamp, ) -from nerve.identity import attribution_for +from nerve.identity import Attribution, attribution_for from tests.actor_rows import FAKE_SYSTEM_ACTOR, mock_system_principal +# A person's actor ID (synthetic), distinct from the system actor. +_ALICE_ID = "00000000-0000-4000-8000-0000000a11ce" + # --------------------------------------------------------------------------- # Helpers @@ -824,13 +827,17 @@ async def test_rotation_schedules_background_memorize(self, cron_service): @pytest.mark.asyncio async def test_rotation_stamps_and_cancels_wakeups(self, cron_service): - """The retired chat is stamped rotated_at, retitled, wakeups cancelled.""" + """The retired chat is stamped rotated_at, retitled, wakeups cancelled. + + The generation has a person as its cause (a "Run now" created it), so + the retitle is Nerve's change for that person. + """ _map_current(cron_service, "cron:pers", { "connected_at": _hours_ago(30), "status": "idle", "title": "Cron: pers", "metadata": "{}", - "caused_by_actor_id": FAKE_SYSTEM_ACTOR.actor_id, + "caused_by_actor_id": _ALICE_ID, }) job = _pers_job(context_rotate_hours=24) @@ -847,7 +854,7 @@ async def test_rotation_stamps_and_cancels_wakeups(self, cron_service): assert title_args[1].startswith("Cron: pers (until ") # The retitle is Nerve's change for the session's cause. assert cron_service.db.update_session_title.call_args.kwargs == { - "attribution": attribution_for(FAKE_SYSTEM_ACTOR, FAKE_SYSTEM_ACTOR), + "attribution": Attribution(FAKE_SYSTEM_ACTOR.actor_id, _ALICE_ID), "reason": "cron_rotation", } diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py index c0562c78d..5f8894c55 100644 --- a/tests/test_schedule_attribution.py +++ b/tests/test_schedule_attribution.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import json from datetime import datetime, timedelta, timezone from types import SimpleNamespace @@ -453,6 +454,67 @@ async def test_run_now_through_rest_runs_as_the_system_for_the_person( assert await install.session_cause_of(turn["session_id"]) == alice +@pytest.mark.asyncio +async def test_a_persistent_generation_that_a_run_now_creates_keeps_the_persons_cause( + install, monkeypatch, +): + """Alice's "Run now" starts a persistent job's first session. + + The session's cause is Alice. A later scheduled run reuses the session + and keeps that cause (a documented approximation). When the session + rotates, its retitle is Nerve's change for Alice, and the next session + is the system actor's. + """ + install.use_fake_runtime(monkeypatch) + alice = install.actor_id("alice") + system = install.system_actor_id + job = CronJob( + id="digest", schedule="1h", prompt="write the digest", + session_mode="persistent", context_rotate_hours=1, + ) + service = _cron_service(install, job) + monkeypatch.setattr("nerve.gateway.server._cron_service", service) + + async with _client(install.app()) as client: + response = await client.post( + "/api/cron/jobs/digest/trigger", headers=install.headers("alice"), + ) + assert response.status_code == 200, response.text + first = install.turns[0]["session_id"] + assert await install.session_cause_of(first) == alice + + await service._run_job_wrapper(job) + assert (install.turns[1]["session_id"], install.turns[1]["cause"]) == (first, alice) + + # The session ages past the rotation window. Generation IDs carry the + # time to the second, so the next one needs a new second. + await install.db.update_session_fields(first, { + "connected_at": (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat(), + }) + await asyncio.sleep(1.1) + await service._run_job_wrapper(job) + + second = install.turns[2]["session_id"] + assert second != first + assert await install.session_cause_of(second) == system + assert install.turns[2]["cause"] == system + async with install.db.db.execute( + "SELECT details, actor_id, caused_by_actor_id FROM session_events" + " WHERE session_id = ? AND event_type = 'title_changed'", + (first,), + ) as cursor: + retitles = [ + (json.loads(row[0]).get("reason"), row[1], row[2]) + async for row in cursor + ] + assert ("cron_rotation", system, alice) in retitles + assert await _cron_logs(install) == [ + ("digest", "success", alice, alice), + ("digest", "success", system, system), + ("digest", "success", system, system), + ] + + @pytest.mark.asyncio async def test_run_now_of_a_workflow_job_starts_the_run_for_the_person( install, monkeypatch, diff --git a/tests/test_wakeups.py b/tests/test_wakeups.py index 7add578fc..f2a71659b 100644 --- a/tests/test_wakeups.py +++ b/tests/test_wakeups.py @@ -175,6 +175,8 @@ async def test_fires_due_wakeup(self, svc): # A self-scheduled loop tick is the session continuing its own work — # it must not re-sort the sidebar. assert kwargs["bump_updated_at"] is False + # The turn runs in its session and takes the session's cause. + assert "cause" not in kwargs # Claimed — no longer pending. assert await svc.db.list_pending_wakeups("s1") == [] @@ -197,6 +199,7 @@ async def test_run_later_wakeup_fires_with_cron_source(self, svc): # The user wrote this message and asked for it to land now, so unlike # a model-scheduled tick it DOES move the session up the sidebar. assert kwargs["bump_updated_at"] is True + assert "cause" not in kwargs @pytest.mark.asyncio async def test_skips_running_session(self, svc): From 77ac43b448e938258c9cfd1b4eca3b463210c075 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Sat, 17 Oct 2026 11:33:10 +0000 Subject: [PATCH 8/9] Give every cron run and generation its own session Isolated cron runs and persistent generations got session IDs with the start time to the second. Two people who used "Run now" on one job in the same second shared one session, so both runs had the cause (and the inference principal) of whoever created it first. An immediate rotation could also reuse the retired generation. A run ID is now the UTC start second and eight random hex digits (new_cron_run_id in nerve/agent/sessions.py), used for isolated runs, persistent generations and run_cron's default. The cron:: prefix stays. The usage report matches the new shape and the older one. Tests: two concurrent "Run now" requests of Alice and Bob in one second get two sessions with their own causes and principal headers; a manual rotation and a rotated generation in the same second get new sessions; the usage report groups both run ID shapes. docs/cron.md and docs/architecture.md describe the run ID. Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 4 +- docs/cron.md | 4 +- nerve/agent/engine.py | 4 +- nerve/agent/sessions.py | 14 +++++ nerve/cron/service.py | 11 ++-- nerve/db/usage.py | 24 +++++--- tests/test_cron.py | 7 ++- tests/test_schedule_attribution.py | 91 ++++++++++++++++++++++++++++-- tests/test_usage_by_cron_job.py | 15 +++++ 9 files changed, 148 insertions(+), 26 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 60517aa71..c751961a0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -71,7 +71,7 @@ Claude Agent SDK wrapper providing: - **Persistent channel mappings** — channel-to-session mapping stored in `channel_sessions` table (survives restarts) - **Orphan recovery** — on startup, sessions marked `active` in DB but with no live client are transitioned to `idle` (resumable) or `stopped` - **Automatic cleanup** — periodic task (every 6h) archives stale sessions (default 30 days) and enforces max session count (default 500) -- **Per-run cron sessions** — each cron run gets a unique session ID (`cron:{job_id}:{timestamp}`) to prevent unbounded message accumulation +- **Per-run cron sessions**: each cron run gets a unique session ID (`cron:{job_id}:{run_id}`, the start time and a random suffix) to prevent unbounded message accumulation - AI-generated session titles via lightweight Haiku API call - Custom MCP tools (tasks, memory recall, conversation history, sync status, skills CRUD, notifications) - API calls routed through configurable base URL (direct Anthropic API or local CLIProxyAPI proxy) @@ -190,7 +190,7 @@ idle-stream support, cache-TTL policy) gate engine behavior — never ### Cron Job 1. APScheduler triggers at schedule -2. Isolated per-run session created (`cron:{job_id}:{timestamp}`) +2. Isolated per-run session created (`cron:{job_id}:{run_id}`) 3. Agent runs with job prompt and cron model, acting as the system principal — the session and the prompt are attributed to it, not to whoever wrote the schedule 4. Agent uses `notify`/`ask_user` tools to communicate with user 5. Run logged in `cron_logs` table diff --git a/docs/cron.md b/docs/cron.md index 71fd6594d..e99e5e95b 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -375,13 +375,13 @@ that name its gate, see below): ### Isolated (default) -Each run creates a fresh session (`cron:{job_id}:{timestamp}`). The agent has no in-context memory of previous runs. This is best for self-contained jobs like daily briefings or cleanup tasks. +Each run creates a fresh session (`cron:{job_id}:{run_id}`; the run ID is the UTC start time `YYYYmmdd-HHMMSS` and eight random hex digits, so two runs that start in the same second get two sessions). The agent has no in-context memory of previous runs. This is best for self-contained jobs like daily briefings or cleanup tasks. ### Persistent Jobs with `session_mode: persistent` maintain SDK conversation context across runs. Each job owns a **generation chat** — one session that is reused run after run until rotation: -- **First trigger**: Creates a fresh generation session (`cron:{job_id}:{timestamp}`) and runs the prompt. +- **First trigger**: Creates a fresh generation session (`cron:{job_id}:{run_id}`, the same ID form as an isolated run) and runs the prompt. - **Subsequent triggers**: Resumes the same SDK session and sends the prompt as a new message. The agent sees all prior runs of this generation in-context. - **Context rotation**: Every `context_rotate_hours` (default: 24) — or daily at `context_rotate_at` — the current chat is **retired and a brand-new chat is started**. The old chat is preserved exactly as it was: it keeps its full message history, stays browsable (and resumable) in the UI like any other session, and ages out via the normal session archival policy. Retiring also indexes the old context into memU, cancels the old session's pending wakeups (so a retired thread can't resurrect itself alongside the new one), and retitles it with its end date (`Cron: {job_id} (until 2026-07-05)`). diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 335c482a2..d1779f7db 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -54,7 +54,7 @@ set_skill_manager, worker_onboarding_prompt, ) -from nerve.agent.sessions import SessionManager, SessionStatus +from nerve.agent.sessions import SessionManager, SessionStatus, new_cron_run_id from nerve.agent.streaming import broadcaster from nerve.agent.tools import ( ToolContext, @@ -3574,7 +3574,7 @@ async def run_cron( a person started it. """ if run_id is None: - run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + run_id = new_cron_run_id() # A cron run is the instance's own work, regardless of who wrote the # schedule or who started the run. A person who started it is the # cause of the run's session. diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 2c9f18531..3bd18b7a8 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -13,6 +13,7 @@ import asyncio import logging +import secrets import uuid from datetime import datetime, timedelta, timezone from enum import StrEnum @@ -30,6 +31,19 @@ logger = logging.getLogger(__name__) + +def new_cron_run_id() -> str: + """A unique run ID for a cron session ID ``cron::``. + + The UTC start time (``YYYYmmdd-HHMMSS``) keeps the IDs in time order; + eight random hex digits make two runs that start in the same second + (for example two people who use "Run now") get two sessions, each with + its own cause. ``nerve/db/usage.py`` matches this shape. + """ + stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + return f"{stamp}-{secrets.token_hex(4)}" + + # Cleanup defaults DEFAULT_ARCHIVE_AFTER_DAYS = 30 DEFAULT_MAX_SESSIONS = 500 diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 5be9bc9ce..c2e4b5150 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -19,6 +19,7 @@ from nerve import paths from nerve.agent.engine import AgentEngine +from nerve.agent.sessions import new_cron_run_id from nerve.config import ConfigError, NerveConfig from nerve.cron.jobs import ( CronJob, @@ -917,8 +918,7 @@ async def _start_new_generation( ``cause`` is the session's cause: by default the system actor; the person whose "Run now" creates the generation. """ - ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") - session_id = f"cron:{job_id}:{ts}" + session_id = f"cron:{job_id}:{new_cron_run_id()}" await self.engine.sessions.get_or_create( session_id, title=f"Cron: {job_id}", source="cron", actor=self.db.system_actor, cause=cause, @@ -1193,10 +1193,9 @@ async def _run_job_inner( ) else: # Isolated mode: per-run session. The run_id is generated - # here (the engine would otherwise generate an identical - # timestamp-based one) so the session id is known for the - # run log. - run_id = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + # here (the engine would otherwise generate one of the same + # shape) so the session id is known for the run log. + run_id = new_cron_run_id() session_id = f"cron:{job.id}:{run_id}" try: await self.db.set_cron_log_session(log_id, session_id) diff --git a/nerve/db/usage.py b/nerve/db/usage.py index a3228fdee..12a8cf09b 100644 --- a/nerve/db/usage.py +++ b/nerve/db/usage.py @@ -36,15 +36,20 @@ # Default fallback when model is unknown or None DEFAULT_PRICING = (5, 25, 0.50, 6.25, 10.00, 0.01) # Opus 4.6 standard -# Cron session ids (see ``SessionManager.create_cron_session`` and -# ``CronService._start_new_generation``): -# cron: — legacy stable persistent session -# cron:: — isolated run / persistent generation -# The run glob matches the second form, so the job id can be cut out of it -# even when the job id itself contains ``:``. -_CRON_RUN_SESSION_GLOB = "cron:?*:" + "[0-9]" * 8 + "-" + "[0-9]" * 6 +# Cron session ids (see ``SessionManager.create_cron_session``, +# ``CronService._start_new_generation`` and ``new_cron_run_id``): +# cron: legacy stable persistent session +# cron::-<8 hex> isolated run / persistent generation +# cron:: the same, from older versions +# The run globs match the last two forms, so the job id can be cut out of +# them even when the job id itself contains ``:``. +_CRON_RUN_STAMP_GLOB = "[0-9]" * 8 + "-" + "[0-9]" * 6 +_CRON_RUN_SESSION_GLOB = "cron:?*:" + _CRON_RUN_STAMP_GLOB + "-" + "[0-9a-f]" * 8 +# len("cron:") + len(":YYYYmmdd-HHMMSS-xxxxxxxx") +_CRON_RUN_SESSION_OVERHEAD = 5 + 25 +_CRON_OLD_RUN_SESSION_GLOB = "cron:?*:" + _CRON_RUN_STAMP_GLOB # len("cron:") + len(":YYYYmmdd-HHMMSS") -_CRON_RUN_SESSION_OVERHEAD = 5 + 16 +_CRON_OLD_RUN_SESSION_OVERHEAD = 5 + 16 def _get_pricing( @@ -317,6 +322,8 @@ async def get_usage_by_cron_job(self, days: int = 7) -> list[dict]: attributed AS ( SELECT CASE WHEN origin_job IS NOT NULL THEN origin_job + WHEN origin_session GLOB ? THEN substr( + origin_session, 6, length(origin_session) - ?) WHEN origin_session GLOB ? THEN substr( origin_session, 6, length(origin_session) - ?) ELSE substr(origin_session, 6) END AS job_id, @@ -362,6 +369,7 @@ async def get_usage_by_cron_job(self, days: int = 7) -> list[dict]: ( window, _CRON_RUN_SESSION_GLOB, _CRON_RUN_SESSION_OVERHEAD, + _CRON_OLD_RUN_SESSION_GLOB, _CRON_OLD_RUN_SESSION_OVERHEAD, window, ), ) as cursor: diff --git a/tests/test_cron.py b/tests/test_cron.py index 2239d49f9..59a1f5562 100644 --- a/tests/test_cron.py +++ b/tests/test_cron.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import re from datetime import datetime, timedelta, timezone from unittest.mock import ANY, AsyncMock, MagicMock, patch from zoneinfo import ZoneInfo @@ -894,11 +895,13 @@ def now(cls, tz=None): return fixed.replace(tzinfo=None) return fixed.astimezone(tz) - with patch("nerve.cron.service.datetime", FixedDateTime): + with patch("nerve.cron.service.datetime", FixedDateTime), \ + patch("nerve.agent.sessions.datetime", FixedDateTime): session_id, rotated = await svc._resolve_persistent_session(job) assert rotated is True - assert session_id == "cron:pers:20260101-150000" + # The run ID is the UTC start second and eight random hex digits. + assert re.fullmatch(r"cron:pers:20260101-150000-[0-9a-f]{8}", session_id) svc.engine.schedule_memorize.assert_awaited_once_with("cron:pers") svc.engine.sessions.mark_idle.assert_not_called() diff --git a/tests/test_schedule_attribution.py b/tests/test_schedule_attribution.py index 5f8894c55..c1dbb3083 100644 --- a/tests/test_schedule_attribution.py +++ b/tests/test_schedule_attribution.py @@ -55,6 +55,20 @@ def _past() -> str: return (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat() +class _OneSecond(datetime): + """Every cron run ID starts in the same second.""" + + @classmethod + def now(cls, tz=None): + return datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc) + + +@pytest.fixture +def same_second(monkeypatch): + """Cron runs and generations that start in one second.""" + monkeypatch.setattr("nerve.agent.sessions.datetime", _OneSecond) + + async def _wakeups(install: _Install, session_id: str) -> list[tuple]: async with install.db.db.execute( "SELECT prompt, status, created_by_actor_id, caused_by_actor_id" @@ -456,7 +470,7 @@ async def test_run_now_through_rest_runs_as_the_system_for_the_person( @pytest.mark.asyncio async def test_a_persistent_generation_that_a_run_now_creates_keeps_the_persons_cause( - install, monkeypatch, + install, monkeypatch, same_second, ): """Alice's "Run now" starts a persistent job's first session. @@ -486,12 +500,11 @@ async def test_a_persistent_generation_that_a_run_now_creates_keeps_the_persons_ await service._run_job_wrapper(job) assert (install.turns[1]["session_id"], install.turns[1]["cause"]) == (first, alice) - # The session ages past the rotation window. Generation IDs carry the - # time to the second, so the next one needs a new second. + # The session ages past the rotation window. The next generation starts + # in the same second as the first one, and still gets its own session. await install.db.update_session_fields(first, { "connected_at": (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat(), }) - await asyncio.sleep(1.1) await service._run_job_wrapper(job) second = install.turns[2]["session_id"] @@ -515,6 +528,37 @@ async def test_a_persistent_generation_that_a_run_now_creates_keeps_the_persons_ ] +@pytest.mark.asyncio +async def test_an_immediate_rotation_starts_a_new_session( + install, monkeypatch, same_second, +): + """A manual rotation right after the first run, in the same second, starts + a new session for the system actor; the first one keeps Alice.""" + install.use_fake_runtime(monkeypatch) + alice = install.actor_id("alice") + job = CronJob( + id="digest", schedule="1h", prompt="write the digest", session_mode="persistent", + ) + service = _cron_service(install, job) + monkeypatch.setattr("nerve.gateway.server._cron_service", service) + + async with _client(install.app()) as client: + await client.post("/api/cron/jobs/digest/trigger", headers=install.headers("alice")) + rotated = (await client.post( + "/api/cron/jobs/digest/rotate", headers=install.headers("alice"), + )).json() + + first = install.turns[0]["session_id"] + assert rotated["rotated"] is True + assert rotated["old_session_id"] == first + assert rotated["new_session_id"] != first + assert rotated["new_session_id"].startswith("cron:digest:20260102-030405-") + assert await install.session_cause_of(first) == alice + assert await install.session_cause_of(rotated["new_session_id"]) == ( + install.system_actor_id + ) + + @pytest.mark.asyncio async def test_run_now_of_a_workflow_job_starts_the_run_for_the_person( install, monkeypatch, @@ -803,3 +847,42 @@ async def test_external_mode_a_run_now_session_names_the_person( assert _named_principals(process) == [BOB_PRINCIPAL] finally: await _stop_idle_watchers(install) + + +@pytest.mark.asyncio +async def test_external_mode_concurrent_run_nows_get_their_own_sessions( + external_install, cli, monkeypatch, same_second, +): + """Alice and Bob use "Run now" on one job at the same time. + + Both runs start in the same second. Each gets its own session, with its + person as cause, and its CLI process names that person. + """ + install = external_install + monkeypatch.setattr(_FakeCli, "install", install) + monkeypatch.delenv(CUSTOM_HEADERS_ENV, raising=False) + job = CronJob(id="nightly", schedule="1h", prompt="tidy the board") + monkeypatch.setattr("nerve.gateway.server._cron_service", _cron_service(install, job)) + alice, bob = _as(ALICE_PRINCIPAL, "Alice"), _as(BOB_PRINCIPAL, "Bob") + try: + async with _client(install.app()) as client: + responses = await asyncio.gather( + client.post("/api/cron/jobs/nightly/trigger", headers=alice), + client.post("/api/cron/jobs/nightly/trigger", headers=bob), + ) + assert [r.status_code for r in responses] == [200, 200] + + sessions = {turn["cause"]: turn["session_id"] for turn in install.turns} + assert set(sessions) == {ALICE_PRINCIPAL, BOB_PRINCIPAL} + assert sessions[ALICE_PRINCIPAL] != sessions[BOB_PRINCIPAL] + for principal, session_id in sessions.items(): + assert session_id.startswith("cron:nightly:20260102-030405-") + assert await install.session_cause_of(session_id) == principal + (process,) = [p for p in cli if p.spec.session_id == session_id] + assert _named_principals(process) == [principal] + assert sorted(await _cron_logs(install)) == sorted([ + ("nightly", "success", ALICE_PRINCIPAL, ALICE_PRINCIPAL), + ("nightly", "success", BOB_PRINCIPAL, BOB_PRINCIPAL), + ]) + finally: + await _stop_idle_watchers(install) diff --git a/tests/test_usage_by_cron_job.py b/tests/test_usage_by_cron_job.py index a4c1d9e9a..76ed7022e 100644 --- a/tests/test_usage_by_cron_job.py +++ b/tests/test_usage_by_cron_job.py @@ -106,6 +106,21 @@ async def test_isolated_runs_roll_up_per_job(self, db: Database): assert row["cost_usd"] == pytest.approx(1.0) assert row["estimated_cost_usd"] == pytest.approx(1.0) + async def test_run_ids_with_a_random_suffix_roll_up_per_job(self, db: Database): + # Run IDs carry the start second and eight random hex digits; the + # older form has no suffix. Both belong to the job, also when the job + # id has colons. + await _turn(db, "cron:alpha:20260101-100000-0a1b2c3d", 0.25) + await _turn(db, "cron:alpha:20260101-100000-ffee0011", 0.50) + await _turn(db, "cron:alpha:20251231-235959", 0.25) + await _turn(db, "cron:team:deploy:20260103-120000-12ab34cd", 0.5) + + rows = _by_job(await db.get_usage_by_cron_job()) + assert set(rows) == {"alpha", "team:deploy"} + assert rows["alpha"]["sessions"] == 3 + assert rows["alpha"]["cost_usd"] == pytest.approx(1.0) + assert rows["team:deploy"]["sessions"] == 1 + async def test_persistent_legacy_and_generation_sessions(self, db: Database): # Legacy stable id and a rotated generation chat both belong to "beta". await _turn(db, "cron:beta", 1.0) From 36520ebf0ef8ce520ee0571267ce0ba83664cd34 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Sat, 17 Oct 2026 11:33:10 +0000 Subject: [PATCH 9/9] Count three session cause approximations in the docs docs/accounts.md lists three approximations of the session cause model (shared sessions, deferred messages and wakeups, a persistent cron session that a person started); the lead sentence said two. Co-Authored-By: Claude Opus 5.5 --- docs/accounts.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/accounts.md b/docs/accounts.md index 996016410..3afd38083 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -234,7 +234,7 @@ session's cause as cause. A change that a person makes directly records that person as both. `messages.actor_id` keeps naming who wrote each message; there is no cause per message or per turn. -This is an approximation in two places: +This is an approximation in three places: - In a session that several people use, the agent's work in every turn is attributed to the person who started the session, also in turns that