diff --git a/docs/accounts.md b/docs/accounts.md index 2c28a37b4..aa7d2c064 100644 --- a/docs/accounts.md +++ b/docs/accounts.md @@ -169,22 +169,74 @@ In external mode: ## Attribution -Sessions store who created them in `created_by_actor_id`. User messages store -who supplied the content in `actor_id`. Both columns contain stable actor IDs; -clients resolve names through `GET /api/actors` — the display name, or the -username when none is set. +Nerve can record two actor IDs for a change: -| Event | Stored actor | +- The **actor** made the change: a person through the web UI, the REST API or + the browser WebSocket, or the system actor when the agent or Nerve did it (a + tool call, a cron job, any internal process). +- The **cause** is the principal on whose behalf the change was made. For a + change that a person made directly, the cause is that person. For work by + the agent, the cause is the cause of its session. + +These columns store actor IDs: + +| Column | What it records | +|---|---| +| `sessions.created_by_actor_id` | Who created the session | +| `sessions.caused_by_actor_id` | The session's cause: on whose behalf all agent work in the session runs | +| `messages.actor_id` | Who supplied the content of a user message | +| `task_events.actor_id`, `task_events.caused_by_actor_id` | Actor and cause of a task status change | +| `task_status_events.actor_id`, `task_status_events.caused_by_actor_id` | Actor and cause of a change to the set of task statuses | + +`nerve/db/actor_columns.py` lists every attribution column. Clients resolve +names through `GET /api/actors`, which returns every actor that these columns +can name: the display name, or the username when none is set. The web UI shows +the system actor with a person as cause as "Nerve for ". + +### The session's cause + +Every session has one cause. Nerve stores it when it creates the session and +never changes it: + +| Session | Cause | |---|---| -| A person creates a session or sends a message | That person's actor | -| Nerve creates a session or prompt, including MCP and Codex sessions | The system actor | -| Telegram, Slack, or imported Codex input | None until a local identity mapping exists | -| Assistant and tool output | None | - -Existing records remain unattributed. Nerve does not infer identity from channel -or source metadata. Account and session administration actions are not yet -recorded as audit events. Missing attribution means legacy history, unidentified -external input, or assistant/tool output; autonomous work uses the system actor. +| 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 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 | +| A workflow leg session | The system actor | +| A Telegram, Slack or other channel session, or an imported Codex session | None until a local identity mapping exists | +| An external MCP client's session | None | + +Everything the agent does in a session (a tool call, a task it changes, a plan +it approves, a session it starts) records the system actor as actor and the +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: + +- 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 + another person sent. +- A deferred message ("run later") or a wakeup that runs in an existing session + is attributed to that session's cause. + +| Work | Actor | Cause | +|---|---|---| +| A person sends a message, or changes a task through the web UI or the REST API | The person | The person | +| The agent works in a session (tool calls, task changes, sessions it starts) | The system actor | The session's cause | +| Telegram, Slack, or imported Codex input | None until a local identity mapping exists | None | +| Assistant and tool output | None | None | + +Existing records remain unattributed: Nerve does not backfill these columns. +Nerve does not infer identity from channel or source metadata. The free-text +origin fields, such as `task_events.actor`, stay as they are; they record where +a change came from, not who made it. Account and session administration actions +are not yet recorded as audit events. Missing attribution means legacy history, +unidentified external input, or assistant/tool output; autonomous work uses the +system actor. ## Session-signing secret diff --git a/docs/api.md b/docs/api.md index d6f8df1fd..8f0d3abb8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -713,9 +713,9 @@ writes no configuration. ### Actors An actor is *who* something is attributed to: a person, or the agent's system -principal. Sessions and messages 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. +principal. Sessions, messages and task events 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. ```json { "id": "…", "kind": "human", "display_name": "Alice", "username": "alice" } @@ -781,13 +781,18 @@ Create a new session. ```json Request: { "title": "My Session" } -Response: { "id": "a1b2c3d4", "title": "My Session", "source": "web", "created_by_actor_id": "…" } +Response: { "id": "a1b2c3d4", "title": "My Session", "source": "web", "created_by_actor_id": "…", "caused_by_actor_id": "…" } ``` `created_by_actor_id` is an actor id for [`GET /api/actors`](#actors), not an account id. It is `null` for legacy rows and sessions caused by an unidentified external person. Every session payload carries the field. +Session payloads also carry `caused_by_actor_id`: the session's cause, on +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). + #### `GET /api/sessions/{id}` Get session details. @@ -996,11 +1001,17 @@ A task's status history, oldest first. ```json Response: { "events": [ - { "id": 1, "task_id": "...", "from_status": null, "to_status": "pending", "actor": "system", "created_at": "..." }, - { "id": 2, "task_id": "...", "from_status": "pending", "to_status": "in_progress", "actor": "impl-abc123", "created_at": "..." } + { "id": 1, "task_id": "...", "from_status": null, "to_status": "pending", "actor": "web", "actor_id": "…", "caused_by_actor_id": "…", "created_at": "..." }, + { "id": 2, "task_id": "...", "from_status": "pending", "to_status": "in_progress", "actor": "impl-abc123", "actor_id": "…", "caused_by_actor_id": "…", "created_at": "..." } ] } ``` +`actor` is the origin of the change (`web`, a session id, `system`, +`backfill`). `actor_id` is who made the change and `caused_by_actor_id` on +whose behalf, both actor ids for [`GET /api/actors`](#actors) and `null` on +rows from before they were recorded. See +[Status History](tasks.md#status-history). + #### `POST /api/tasks/{id}/move` Place a task in a lane — the board's drag-and-drop write. Send *intent* ("between these two cards"), not a computed rank: the server resolves the diff --git a/docs/tasks.md b/docs/tasks.md index 4a571f75f..4775c027c 100644 --- a/docs/tasks.md +++ b/docs/tasks.md @@ -119,8 +119,9 @@ Two rules keep ranks meaningful: ### Status History Every status transition is appended to `task_events` (`task_id`, -`from_status`, `to_status`, `actor`, `created_at`), added in migration v044 -and seeded with one origin row per existing task. `from_status` is NULL on +`from_status`, `to_status`, `actor`, `actor_id`, `caused_by_actor_id`, +`created_at`), added in migration v044 and seeded with one origin row per +existing task. `from_status` is NULL on the row recording a task's creation, so an aging calculation can tell "created here" from "moved here". @@ -134,11 +135,28 @@ rewrites every row with the status it already had. The one case where reindex *does* change a status, resetting an orphaned row to match its directory, is a real correction and gets a row. -`actor` is the session id for agent-driven changes, `web` for the HTTP API, -`backfill` on the origin rows v044 seeded for tasks that predate the table, -and `system` otherwise. `backfill` is load-bearing rather than cosmetic: a -real creation also records a NULL `from_status`, so the actor is the only -thing distinguishing a synthesized origin from one that actually happened. +`actor` is the origin of the change, not an identity: the session id for +agent-driven changes, `web` for the HTTP API, `backfill` on the origin rows +v044 seeded for tasks that predate the table, and `system` otherwise. +`backfill` is load-bearing rather than cosmetic: a real creation also records +a NULL `from_status`, so the actor is the only thing distinguishing a +synthesized origin from one that actually happened. + +`actor_id` and `caused_by_actor_id` (migration v053) name who made the change +and on whose behalf, as actor IDs (see +[Attribution](accounts.md#attribution)). A change through the HTTP API records +the person as both. A change by an agent tool call records the system actor +and the cause of its session: the person who started the session, or the +system actor for a cron job. Rows written before v053 have NULL in both +columns. The `GET /api/tasks/{id}/events` rows include both columns, and the +timeline shows "Nerve for " when the agent changed the status in a +session that the person started. + +Changes to the set of statuses (create, update, reorder, delete, through the +HTTP API or the `task_status_create` tool) append a row to +`task_status_events` (`change`, `status_name`, `detail`, `actor_id`, +`caused_by_actor_id`, `created_at`), added in migration v053. `status_name` is +NULL for a reorder, and `detail` is a JSON object with the new values. This powers the aging indicator on board cards and the timeline in the task detail view. Tasks whose last transition predates v044 report no entry time @@ -151,6 +169,14 @@ Every write to a task row advances `tasks.revision` by one, added in migration v052. The row routes return it, and `task_read` ends with a `` line that `task_write` strips again. +Changes of one task that write its file and its row run one at a time: the +`PATCH` and `move` routes, `task_update`, `task_write`, `task_done` and the +reopen path hold a lock per task from their first read of the row to their +last write (`nerve/tasks/locks.py`). So an edit always lands in the file that +the row names, also when another change moves the file into or out of +`done/` at the same time. Changes of different tasks do not wait for each +other. + Pass the value you read as `expect_revision` to `task_update` or `task_done` to make the write conditional: if the row has changed since, nothing is written and the tool reports a conflict, so re-read and retry. Without it, diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 8dae74764..8981e8350 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -69,7 +69,7 @@ from nerve.agent.tools import init_tools from nerve.config import NerveConfig, RESUME_QUEUE_FILE, load_mcp_servers from nerve.db import Database -from nerve.identity import Actor +from nerve.identity import CAUSE_FROM_ACTOR, Actor, CauseArg, resolve_cause from nerve.observability.langfuse import attributes as lf_attrs from nerve.setup_writer import ( TASK_ONBOARDING_MARKER, @@ -375,12 +375,28 @@ def get_active_channel(self, session_id: str) -> str | None: """Return the channel name currently driving ``session_id`` (or None).""" return self._active_channel.get(session_id) + def session_cause(self, session_id: str) -> str | None: + """The session's stored cause: who all agent work in it runs for. + + That is the person who started the session, the system actor for a + session the instance started for itself, or ``None`` when unknown + (channel input with no known person, an imported or MCP satellite + session). The value is set once when the session is created and never + changes. See :meth:`nerve.db.sessions.SessionStore.session_cause`. + """ + return self.db.session_cause(session_id) + # ------------------------------------------------------------------ # # Backend resolution (sticky per session) # # ------------------------------------------------------------------ # def _build_tool_context(self, session_id: str) -> ToolContext: - """Fresh per-session ToolContext for backend MCP servers.""" + """Fresh per-session ToolContext for backend MCP servers. + + The agent's tool calls are the system actor's changes, made for the + session's stored cause. The value never changes, so the context + keeps the one read here (the client build has just read the row). + """ return ToolContext( session_id=session_id, workspace=self.config.workspace, @@ -391,6 +407,8 @@ def _build_tool_context(self, session_id: str) -> ToolContext: skill_manager=self._skill_manager, engine=self, notification_service=self.notification_service, + actor=self.db.system_actor, + cause=self.session_cause(session_id), ) def _gateway_port(self) -> int | None: @@ -2494,6 +2512,7 @@ async def run( bump_updated_at: bool = True, *, actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> str: """Run the agent for a user message and return the final text response. @@ -2507,6 +2526,11 @@ async def run( reply, whose authorship is its ``role``. Required keyword-only: an autonomous caller must say so rather than inherit a person's identity by omission. + cause: The session's cause, used only when this run creates the + session: an :class:`Actor`, an actor ID, or ``None`` + for unknown. The default is the actor. An existing + session keeps its stored cause (see + :meth:`session_cause`), whoever sends the message. internal: If True, the user_message is a system-generated trigger (e.g., background task completion) and won't be stored in DB or shown in the UI. @@ -2523,6 +2547,9 @@ async def run( trigger message", and plenty of internal turns (a run-later deferral, the star hook) are user-requested. """ + # Resolved before the lock: an invalid cause fails here, before the + # session is marked as running. + cause_id = resolve_cause(actor, cause) # Serialize runs per session — messages for the same session wait # in order instead of failing with "already running". lock = self._session_locks.setdefault(session_id, asyncio.Lock()) @@ -2554,6 +2581,7 @@ async def run( image_refs=image_refs, bump_updated_at=bump_updated_at, actor=actor, + cause_id=cause_id, ) finally: self.sessions.mark_not_running(session_id) @@ -2600,9 +2628,13 @@ async def _run_inner( bump_updated_at: bool = True, *, actor: Actor | None, + cause_id: str | None, ) -> str: - # Ensure session exists in DB - await self.sessions.get_or_create(session_id, source=source, actor=actor) + # Ensure session exists in DB. ``cause_id`` applies only if the row + # is created here. + await self.sessions.get_or_create( + session_id, source=source, actor=actor, cause=cause_id, + ) session = await self.db.get_session(session_id) diff --git a/nerve/agent/plan_service.py b/nerve/agent/plan_service.py index a8dc12c08..520b3083e 100644 --- a/nerve/agent/plan_service.py +++ b/nerve/agent/plan_service.py @@ -22,6 +22,7 @@ if TYPE_CHECKING: from nerve.agent.engine import AgentEngine from nerve.db import Database + from nerve.identity import ActorOrId logger = logging.getLogger(__name__) @@ -57,6 +58,8 @@ async def request_plan_revision( engine: "AgentEngine", plan_id: str, feedback: str, + *, + cause: "ActorOrId | None", ) -> dict: """Persist revision feedback and dispatch a revision prompt to the planner. @@ -65,6 +68,10 @@ async def request_plan_revision( engine: AgentEngine instance (used to ensure session + dispatch run). plan_id: The pending plan to revise. feedback: Free-text feedback explaining what should change. + cause: Who asked for the revision: the person for the HTTP route, + the calling session's cause for the tool. It becomes the cause of + the planner session if this call creates it; an existing planner + session keeps its own cause. Returns: ``{"plan_id", "task_id", "session_id", "status"}`` on success. @@ -146,7 +153,7 @@ async def request_plan_revision( ) await engine.sessions.get_or_create( session_id, title=session_title, source="cron", - actor=engine.db.system_actor, + actor=engine.db.system_actor, cause=cause, ) asyncio.create_task( engine.run( diff --git a/nerve/agent/sessions.py b/nerve/agent/sessions.py index 7a34d80ba..d192712b6 100644 --- a/nerve/agent/sessions.py +++ b/nerve/agent/sessions.py @@ -19,7 +19,7 @@ from typing import Any from nerve.db import Database -from nerve.identity import Actor +from nerve.identity import CAUSE_FROM_ACTOR, Actor, CauseArg logger = logging.getLogger(__name__) @@ -101,19 +101,21 @@ async def get_or_create( cwd: str | None = None, *, actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> dict: """Get an existing session or create a new one. - ``actor`` is only used when a row is actually created — an existing - session keeps the creator it was stamped with, so re-resolving a - session someone else started never re-attributes it. Required - keyword-only, like the store method underneath it. + ``actor`` and ``cause`` are only used when a row is actually created. + An existing session keeps the creator and the cause it was stamped + with, so re-resolving a session someone else started never + re-attributes it. ``actor`` is required keyword-only, like the store + method underneath it; ``cause`` defaults to the actor. """ session = await self.db.get_session(session_id) if not session: session = await self._create_session( session_id, title=title, source=source, metadata=metadata, - backend=backend, model=model, cwd=cwd, actor=actor, + backend=backend, model=model, cwd=cwd, actor=actor, cause=cause, ) return session @@ -130,6 +132,7 @@ async def _create_session( cwd: str | None = None, *, actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> dict: """Create a new session with status=created and log the event.""" if backend is None: @@ -154,6 +157,7 @@ async def _create_session( model=model, cwd=cwd, actor=actor, + cause=cause, ) await self.db.log_session_event(session_id, "created", { "source": source, @@ -649,7 +653,7 @@ async def add_message( """Add a message to a session's history. ``actor`` is the principal whose input this row records, or ``None`` - for assistant and tool output — see + for assistant and tool output. See :meth:`nerve.db.messages.MessageStore.add_message`. ``bump_updated_at=False`` records the message without moving the diff --git a/nerve/agent/tools/handlers/plans.py b/nerve/agent/tools/handlers/plans.py index ae231dc29..f151ba06c 100644 --- a/nerve/agent/tools/handlers/plans.py +++ b/nerve/agent/tools/handlers/plans.py @@ -227,8 +227,10 @@ async def plan_approve_handler(ctx: ToolContext, args: dict) -> ToolResult: plan_type = plan.get("plan_type", "generic") # The agent approved this plan through its own tool, so both the session - # and the run it dispatches are the instance's own work. + # and the run it dispatches are the instance's own work. The session's + # cause is the cause of the session in which the agent approved it. impl_actor = ctx.db.system_actor + impl_cause = ctx.cause_id # Mark as implementing (prevents double-approve) await ctx.db.update_plan(plan_id, status="implementing", reviewed_at=now) @@ -236,7 +238,7 @@ async def plan_approve_handler(ctx: ToolContext, args: dict) -> ToolResult: impl_session_id = f"impl-{str(uuid.uuid4())[:8]}" await ctx.engine.sessions.get_or_create( impl_session_id, title=f"Implement: {task['title']}", source="web", - actor=impl_actor, + actor=impl_actor, cause=impl_cause, ) await ctx.db.update_plan(plan_id, impl_session_id=impl_session_id) @@ -381,6 +383,7 @@ async def plan_revise_handler(ctx: ToolContext, args: dict) -> ToolResult: try: result = await request_plan_revision( db=ctx.db, engine=ctx.engine, plan_id=plan_id, feedback=feedback, + cause=ctx.cause_id, ) except PlanNotFound: return ToolResult.text(f"Plan not found: {plan_id}") diff --git a/nerve/agent/tools/handlers/tasks.py b/nerve/agent/tools/handlers/tasks.py index 223bb1f06..d3338abcd 100644 --- a/nerve/agent/tools/handlers/tasks.py +++ b/nerve/agent/tools/handlers/tasks.py @@ -12,6 +12,7 @@ from __future__ import annotations import asyncio +import functools import logging import re from datetime import datetime, timezone @@ -31,6 +32,7 @@ TASK_WRITE_SCHEMA, ) from nerve.config import ensure_path_not_tracked_config +from nerve.db.tasks import KEEP from nerve.db.task_statuses import ( DEFAULT_STATUS, STATUS_NAME_RE, @@ -39,10 +41,26 @@ random_status_color, ) from nerve.tasks.files import move_task_file, write_task_file +from nerve.tasks.locks import task_mutation logger = logging.getLogger(__name__) +def _one_change_per_task(handler): + """Run a handler that writes a task file and its row under the task's lock. + + See :mod:`nerve.tasks.locks`. The lock covers the handler's first read + of the row, so the path that it writes is the path the row names. + """ + + @functools.wraps(handler) + async def wrapper(ctx: ToolContext, args: dict) -> ToolResult: + async with task_mutation(args["task_id"]): + return await handler(ctx, args) + + return wrapper + + async def _format_status_reminder(ctx: ToolContext, header: str) -> str: """Build a reminder listing every configured status + its description.""" statuses = await ctx.db.list_task_statuses() if ctx.db else [] @@ -357,6 +375,7 @@ async def task_create_handler(ctx: ToolContext, args: dict) -> ToolResult: tags=tags_to_string(tags), content=content, actor=ctx.session_id, + attribution=ctx.attribution, ) except Exception: # A claimed ID is now sticky, so a half-done create would burn it: @@ -410,6 +429,7 @@ async def task_list_handler(ctx: ToolContext, args: dict) -> ToolResult: return ToolResult.text("\n".join(lines)) +@_one_change_per_task async def task_update_handler(ctx: ToolContext, args: dict) -> ToolResult: from nerve.tasks.models import parse_tags_string, tags_to_string @@ -529,7 +549,6 @@ async def task_update_handler(ctx: ToolContext, args: dict) -> ToolResult: content = re.sub(r"\*\*Tags:\*\* .*\n?", "", content, count=1) final_title = new_title or task["title"] - final_status = status or task["status"] # Presence carries through to the row the same way it does to # the file above: an omitted column keeps its stored value, a # present-but-empty one clears it. @@ -539,9 +558,14 @@ async def task_update_handler(ctx: ToolContext, args: dict) -> ToolResult: if raw_tags is not None: edits["tags"] = new_tags_str write = lambda: write_task_file(file_path, content) + # KEEP for what this call does not change: the path belongs to + # task_done / task_reopen, and a status that another writer + # set after the read above must stay. A requested status is + # applied in the write's transaction, from the stored one. kw = dict( - task_id=task_id, file_path=task["file_path"], title=final_title, - status=final_status, content=content, actor=ctx.session_id, **edits, + task_id=task_id, file_path=KEEP, title=final_title, + status=status or KEEP, content=content, actor=ctx.session_id, + attribution=ctx.attribution, **edits, ) # Without a token the order stays file-then-row, as before. With # one the row is written first, so a refused caller writes no file. @@ -574,6 +598,7 @@ async def task_update_handler(ctx: ToolContext, args: dict) -> ToolResult: if status: applied = await ctx.db.transition_task( task_id, status, expect_revision=expect_revision, actor=ctx.session_id, + attribution=ctx.attribution, **({"tags": new_tags_str} if raw_tags is not None else {}), ) elif raw_tags is not None: @@ -618,6 +643,7 @@ async def task_read_handler(ctx: ToolContext, args: dict) -> ToolResult: return ToolResult.text(f"Task file not found for: {task_id}") +@_one_change_per_task async def task_write_handler(ctx: ToolContext, args: dict) -> ToolResult: task_id = args["task_id"] new_content = _REVISION_TRAILER_RE.sub("", args.get("content", "")) @@ -661,13 +687,17 @@ async def task_write_handler(ctx: ToolContext, args: dict) -> ToolResult: if "tags" in frontmatter: edits["tags"] = tags_to_string(parse_tags_string(frontmatter["tags"])) + # task_write changes neither the path nor the status. KEEP reads both in + # the write's transaction, so a move by another writer since the read + # above stays. await ctx.db.upsert_task( task_id=task_id, - file_path=task["file_path"], + file_path=KEEP, title=new_title, - status=task["status"], + status=KEEP, deadline=new_deadline or None, content=new_content, + attribution=ctx.attribution, **edits, ) @@ -675,6 +705,7 @@ async def task_write_handler(ctx: ToolContext, args: dict) -> ToolResult: return ToolResult.text(f"Task {task_id} written ({len(new_content)} chars).") +@_one_change_per_task async def task_done_handler(ctx: ToolContext, args: dict) -> ToolResult: task_id = args["task_id"] note = args.get("note", "") @@ -711,19 +742,21 @@ async def task_done_handler(ctx: ToolContext, args: dict) -> ToolResult: if expect_revision is None: await asyncio.to_thread(move_task_file, src, dst, content) await ctx.db.transition_task( - task_id, "done", actor=ctx.session_id, + task_id, "done", actor=ctx.session_id, attribution=ctx.attribution, file_path=rel_path, content=content, ) else: if not await ctx.db.transition_task( task_id, "done", expect_revision=expect_revision, - actor=ctx.session_id, file_path=rel_path, content=content, + actor=ctx.session_id, attribution=ctx.attribution, + file_path=rel_path, content=content, ): return _revision_conflict(task_id, expect_revision) await asyncio.to_thread(move_task_file, src, dst, content) else: done = await ctx.db.transition_task( task_id, "done", expect_revision=expect_revision, actor=ctx.session_id, + attribution=ctx.attribution, ) if not done and expect_revision is not None: return _revision_conflict(task_id, expect_revision) @@ -739,6 +772,7 @@ async def task_done_handler(ctx: ToolContext, args: dict) -> ToolResult: return ToolResult.text(f"Task {task_id} marked as done.") +@_one_change_per_task async def task_reopen_handler(ctx: ToolContext, args: dict) -> ToolResult: """Mirror of ``task_done``: bring a completed task back to life. @@ -803,6 +837,7 @@ async def task_reopen_handler(ctx: ToolContext, args: dict) -> ToolResult: kw: dict = dict( expect=(TERMINAL_STATUS,), expect_revision=expect_revision, actor=ctx.session_id, + attribution=ctx.attribution, ) if src is not None: content = await asyncio.to_thread(src.read_text, encoding="utf-8") @@ -865,6 +900,7 @@ async def task_status_create_handler(ctx: ToolContext, args: dict) -> ToolResult created = await ctx.db.create_task_status( name=name, label=label, color=color, description=description, + attribution=ctx.attribution, ) return ToolResult.text( f"Created task status '{created['name']}' " diff --git a/nerve/agent/tools/registry.py b/nerve/agent/tools/registry.py index f6078c383..96f3e276d 100644 --- a/nerve/agent/tools/registry.py +++ b/nerve/agent/tools/registry.py @@ -18,6 +18,8 @@ from pathlib import Path from typing import Any, Awaitable, Callable, TYPE_CHECKING +from nerve.identity import Actor, ActorOrId, Attribution, actor_id_of + if TYPE_CHECKING: from nerve.agent.engine import AgentEngine from nerve.config import NerveConfig @@ -39,6 +41,14 @@ class ToolContext: Fields are concrete types rather than Protocols to keep call sites readable; tests construct ``ToolContext`` with mocks/None for fields a given handler doesn't need. + + ``actor`` performs the tool call: the system actor when the agent calls a + tool, the person when an HTTP route invokes a handler. ``cause`` is the + principal on whose behalf the call happens: for an agent tool call, the + session's stored cause (``sessions.caused_by_actor_id``); for a route, the + person. It is a fixed value, or a callable that returns the cause when a + handler reads it. Read it through :attr:`cause_id` or :attr:`attribution`, + never directly. """ session_id: str @@ -51,6 +61,21 @@ class ToolContext: engine: "AgentEngine | None" = None notification_service: "NotificationService | None" = None runtime_metadata: dict[str, Any] = field(default_factory=dict) + actor: Actor | None = None + cause: "ActorOrId | Callable[[], ActorOrId | None] | None" = None + + @property + def cause_id(self) -> str | None: + """The cause actor ID at the time of the read, or ``None`` if unknown.""" + cause = self.cause() if callable(self.cause) else self.cause + return actor_id_of(cause) + + @property + def attribution(self) -> Attribution: + """The pair that a write made by this tool call records.""" + return Attribution( + actor_id=actor_id_of(self.actor), caused_by_actor_id=self.cause_id, + ) @dataclass diff --git a/nerve/db/actor_columns.py b/nerve/db/actor_columns.py new file mode 100644 index 000000000..99a18eda8 --- /dev/null +++ b/nerve/db/actor_columns.py @@ -0,0 +1,22 @@ +"""Every attribution column that stores an ``actor_refs`` ID. + +One ``(table, column)`` pair per line. A migration that adds an attribution +column appends its pair at the end. The tests compare this list with the +schema: each pair must be a column with a foreign key to ``actor_refs``, and +each such column must be in the list, except ``accounts.actor_id``, which +binds a login to its person and is not attribution. + +``GET /api/actors`` must return every actor that one of these columns names. +""" + +from __future__ import annotations + +ACTOR_COLUMNS: tuple[tuple[str, str], ...] = ( + ("sessions", "created_by_actor_id"), + ("messages", "actor_id"), + ("task_events", "actor_id"), + ("task_events", "caused_by_actor_id"), + ("sessions", "caused_by_actor_id"), + ("task_status_events", "actor_id"), + ("task_status_events", "caused_by_actor_id"), +) diff --git a/nerve/db/base.py b/nerve/db/base.py index 39505dd4a..10c29f7e3 100644 --- a/nerve/db/base.py +++ b/nerve/db/base.py @@ -341,6 +341,10 @@ def __init__(self, db_path: Path, workspace: Path | None = None): # a caller or test can tune them before connect() (e.g. busy_timeout=0). self._pragmas: dict[str, object] = dict(_DEFAULT_PRAGMAS) self._system_actor: Actor | None = None + # Each session's stored cause (sessions.caused_by_actor_id). Loaded at + # connect and kept by create_session, get_session and delete_session; + # a session's cause never changes. Read through session_cause(). + self._session_causes: dict[str, str | None] = {} # Actor ID to the last display name upsert_external_actor() wrote. self._external_actor_names: dict[str, str | None] = {} # The state-file modes connect() found after its repair. The identity @@ -417,6 +421,7 @@ async def connect(self) -> None: ) await run_migrations(self._db) await self._cache_system_actor() + await self._cache_session_causes() # After migrations, so the table exists. A key that other users # could read is compromised. if exposed: @@ -425,6 +430,7 @@ async def connect(self) -> None: except BaseException: db, self._db = self._db, None self._system_actor = None + self._session_causes = {} try: await db.close() except Exception as e: # noqa: BLE001 — never mask the real failure @@ -515,6 +521,7 @@ async def close(self) -> None: await self._db.close() self._db = None self._system_actor = None + self._session_causes = {} self._external_actor_names.clear() @property diff --git a/nerve/db/migrations/v053_cause_attribution.py b/nerve/db/migrations/v053_cause_attribution.py new file mode 100644 index 000000000..c2c8bb0cf --- /dev/null +++ b/nerve/db/migrations/v053_cause_attribution.py @@ -0,0 +1,85 @@ +"""V53: record the cause of a session and who changes tasks and task statuses. + +Adds three nullable columns that hold an ``actor_refs.id``: + +- ``sessions.caused_by_actor_id``: the session's cause, the principal on + whose behalf all agent work in the session runs. It is set once when the + session is created and never changes: the person who started the session, + the system actor for a session that the instance starts for itself, or + ``NULL`` when unknown (channel input with no known person, an imported or + MCP satellite session). +- ``task_events.actor_id``: who changed the status. A person through the UI or + REST, or the system actor for an agent tool call or another internal + process. +- ``task_events.caused_by_actor_id``: on whose behalf the change was made. + +``task_events.actor`` stays. It is the origin of the change (``web``, a +session ID, ``system``, ``backfill``), not an identity. + +Adds the table ``task_status_events``: one append-only row for each change to +the set of task statuses (``created``, ``updated``, ``reordered``, +``deleted``), with its actor and cause. A status definition row can be +deleted, so its history cannot live on the row. + +**Existing rows stay NULL and nothing is backfilled.** NULL reads as "not +recorded". The free-text origin of an old task event or the creator of an old +session does not prove a cause, so a backfill would invent attribution. + +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 = ( + ("sessions", "caused_by_actor_id", "TEXT REFERENCES actor_refs(id)"), + ("task_events", "actor_id", "TEXT REFERENCES actor_refs(id)"), + ("task_events", "caused_by_actor_id", "TEXT REFERENCES actor_refs(id)"), +) + +# ``status_name`` is NULL for ``reordered``, which changes the order of all +# statuses; ``detail`` is a JSON object with the changed values. +_STATUS_EVENTS_SQL = """ +CREATE TABLE IF NOT EXISTS task_status_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + change TEXT NOT NULL, + status_name TEXT, + detail TEXT, + actor_id TEXT REFERENCES actor_refs(id), + caused_by_actor_id TEXT REFERENCES actor_refs(id), + created_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_task_status_events_created + ON task_status_events(created_at); +""" + + +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}") + await db.executescript(_STATUS_EVENTS_SQL) + logger.info( + "v053: cause and task attribution columns %s; task_status_events ready", + ", ".join(added) if added else "already present", + ) diff --git a/nerve/db/sessions.py b/nerve/db/sessions.py index e20941ab8..b55a7297e 100644 --- a/nerve/db/sessions.py +++ b/nerve/db/sessions.py @@ -5,7 +5,7 @@ import json from datetime import datetime, timezone -from nerve.identity import Actor +from nerve.identity import CAUSE_FROM_ACTOR, Actor, CauseArg, resolve_cause # Sources the sidebar treats as "system" (machine-driven); everything else is a conversation by exclusion, so a new source shows up in the feed by default. SYSTEM_SOURCES = ("cron", "hook") @@ -29,6 +29,7 @@ async def create_session( cwd: str | None = None, *, actor: Actor | None, + cause: CauseArg = CAUSE_FROM_ACTOR, ) -> dict: """Insert a session row, recording who caused it to exist. @@ -39,8 +40,14 @@ async def create_session( sessions the instance mints for itself, and ``None`` where attribution is unavailable. ``source`` is transport provenance, never identity. - The ``OR IGNORE`` insert makes creator attribution write-once; - re-resolving an existing session cannot replace it. + ``cause`` is the session's cause (``caused_by_actor_id``): the + principal on whose behalf all agent work in the session runs. The + default is the actor. Pass it when they differ, for example ``None`` + for a session that the system actor creates for input from an unknown + person. + + The ``OR IGNORE`` insert makes creator and cause attribution + write-once; re-resolving an existing session cannot replace them. Returns the **stored** row, not the arguments. The distinction only shows up when the insert was ignored — a same-id race, or a short-id @@ -50,16 +57,17 @@ async def create_session( """ now = datetime.now(timezone.utc).isoformat() created_by = actor.actor_id if actor else None + caused_by = resolve_cause(actor, cause) await self._write( """INSERT OR IGNORE INTO sessions (id, title, source, metadata, status, parent_session_id, forked_from_message, backend, model, cwd, created_by_actor_id, - created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + caused_by_actor_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (session_id, title or session_id, source, json.dumps(metadata or {}), status, parent_session_id, forked_from_message, backend, model, cwd, - created_by, now, now), + created_by, caused_by, now, now), ) stored = await self.get_session(session_id) if stored is not None: @@ -72,12 +80,38 @@ async def create_session( "parent_session_id": parent_session_id, "backend": backend, "model": model, "cwd": cwd, "created_by_actor_id": created_by, + "caused_by_actor_id": caused_by, } async def get_session(self, session_id: str) -> dict | None: async with self.db.execute("SELECT * FROM sessions WHERE id = ?", (session_id,)) as cursor: row = await cursor.fetchone() - return dict(row) if row else None + if not row: + return None + session = dict(row) + # A row that another connection wrote after connect() is known from + # its first read on. + self._session_causes[session_id] = session.get("caused_by_actor_id") + return session + + async def _cache_session_causes(self) -> None: + """Load the stored cause of every session. Called by ``connect()``.""" + async with self.db.execute( + "SELECT id, caused_by_actor_id FROM sessions" + ) as cursor: + self._session_causes = {row[0]: row[1] async for row in cursor} + + def session_cause(self, session_id: str) -> str | None: + """The stored cause of a session: who all agent work in it runs for. + + Read from memory, without a query: the value is loaded for every + session at connect, and ``create_session`` and ``get_session`` keep it + for the rows they write and read. It never changes. ``None`` is a + stored NULL (no known cause) or a session that this connection does + not know: one that does not exist, or that another connection created + and this one has not read yet. + """ + return self._session_causes.get(session_id) async def list_sessions( self, limit: int = 50, include_archived: bool = False, @@ -256,6 +290,7 @@ async def delete_session(self, session_id: str) -> None: await self.db.execute("DELETE FROM channel_sessions WHERE session_id = ?", (session_id,)) await self.db.execute("DELETE FROM messages WHERE session_id = ?", (session_id,)) await self.db.execute("DELETE FROM sessions WHERE id = ?", (session_id,)) + self._session_causes.pop(session_id, None) async def update_session_metadata(self, session_id: str, metadata: dict) -> None: """Update the metadata JSON for a session. diff --git a/nerve/db/task_statuses.py b/nerve/db/task_statuses.py index c77148b3d..ed5333d74 100644 --- a/nerve/db/task_statuses.py +++ b/nerve/db/task_statuses.py @@ -8,10 +8,13 @@ from __future__ import annotations +import json import random import re from datetime import datetime, timezone +from nerve.identity import UNATTRIBUTED, Attribution + # Status names are slug-like so they're safe to store on tasks.status and # render without escaping: lowercase alphanumerics + underscores, starting # with an alphanumeric. @@ -59,7 +62,39 @@ def normalize_color(color: str | None) -> str: class TaskStatusStore: - """Mixin providing CRUD for the configurable ``task_statuses`` table.""" + """Mixin providing CRUD for the configurable ``task_statuses`` table. + + Every change also appends a row to ``task_status_events`` in the same + transaction, with the ``attribution`` that the caller passes. + """ + + async def _record_status_def_event( + self, + change: str, + status_name: str | None, + detail: dict | None, + attribution: Attribution, + ) -> None: + """Append a status definition change. Caller must hold ``_atomic()``.""" + await self.db.execute( + """INSERT INTO task_status_events + (change, status_name, detail, actor_id, caused_by_actor_id, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + ( + change, status_name, + json.dumps(detail, sort_keys=True) if detail else None, + attribution.actor_id, attribution.caused_by_actor_id, + datetime.now(timezone.utc).isoformat(), + ), + ) + + async def list_task_status_events(self, limit: int = 200) -> list[dict]: + """Status definition changes, oldest first.""" + async with self.db.execute( + "SELECT * FROM task_status_events ORDER BY created_at ASC, id ASC LIMIT ?", + (limit,), + ) as cursor: + return [dict(row) async for row in cursor] async def list_task_statuses(self) -> list[dict]: """Return all status definitions ordered by sort_order then name.""" @@ -68,7 +103,9 @@ async def list_task_statuses(self) -> list[dict]: ) as cursor: return [dict(row) async for row in cursor] - async def reorder_task_statuses(self, names: list[str]) -> list[dict]: + async def reorder_task_statuses( + self, names: list[str], *, attribution: Attribution = UNATTRIBUTED, + ) -> list[dict]: """Rewrite ``sort_order`` to match the given sequence of names. Takes the whole desired order rather than one status's new index: @@ -93,6 +130,9 @@ async def reorder_task_statuses(self, names: list[str]) -> list[dict]: "UPDATE task_statuses SET sort_order = ? WHERE name = ?", [(i, name) for i, name in enumerate(ordered)], ) + await self._record_status_def_event( + "reordered", None, {"order": ordered}, attribution, + ) return await self.list_task_statuses() @@ -117,6 +157,8 @@ async def create_task_status( color: str | None = None, description: str = "", is_system: int = 0, + *, + attribution: Attribution = UNATTRIBUTED, ) -> dict: """Insert a new status definition. Caller validates name uniqueness.""" now = datetime.now(timezone.utc).isoformat() @@ -124,14 +166,20 @@ async def create_task_status( "SELECT COALESCE(MAX(sort_order), -1) FROM task_statuses" ) as cur: next_order = (await cur.fetchone())[0] + 1 + color = normalize_color(color) async with self._atomic(): await self.db.execute( """INSERT INTO task_statuses (name, label, color, description, is_system, sort_order, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", - (name, label, normalize_color(color), description, + (name, label, color, description, 1 if is_system else 0, next_order, now), ) + await self._record_status_def_event( + "created", name, + {"label": label, "color": color, "description": description}, + attribution, + ) return await self.get_task_status_def(name) # type: ignore[return-value] async def update_task_status_def( @@ -142,33 +190,43 @@ async def update_task_status_def( color: str | None = None, description: str | None = None, sort_order: int | None = None, + attribution: Attribution = UNATTRIBUTED, ) -> None: """Patch mutable fields of a status definition (never the name).""" - sets: list[str] = [] - params: list = [] + changes: dict = {} if label is not None: - sets.append("label = ?") - params.append(label) + changes["label"] = label if color is not None: - sets.append("color = ?") - params.append(normalize_color(color)) + changes["color"] = normalize_color(color) if description is not None: - sets.append("description = ?") - params.append(description) + changes["description"] = description if sort_order is not None: - sets.append("sort_order = ?") - params.append(sort_order) - if not sets: + changes["sort_order"] = sort_order + if not changes: return - params.append(name) - await self._write( - f"UPDATE task_statuses SET {', '.join(sets)} WHERE name = ?", - tuple(params), - ) + async with self._atomic(): + cursor = await self.db.execute( + f"UPDATE task_statuses SET {', '.join(f'{c} = ?' for c in changes)} " + "WHERE name = ?", + (*changes.values(), name), + ) + updated = (cursor.rowcount or 0) == 1 + await cursor.close() + if updated: + await self._record_status_def_event("updated", name, changes, attribution) - async def delete_task_status_def(self, name: str) -> None: + async def delete_task_status_def( + self, name: str, *, attribution: Attribution = UNATTRIBUTED, + ) -> None: """Delete a status definition. Caller enforces protection rules.""" - await self._write("DELETE FROM task_statuses WHERE name = ?", (name,)) + async with self._atomic(): + cursor = await self.db.execute( + "DELETE FROM task_statuses WHERE name = ?", (name,), + ) + deleted = (cursor.rowcount or 0) == 1 + await cursor.close() + if deleted: + await self._record_status_def_event("deleted", name, None, attribution) async def count_tasks_with_status(self, name: str) -> int: """Count tasks currently set to the given status.""" diff --git a/nerve/db/tasks.py b/nerve/db/tasks.py index bad969ff1..2e58fa379 100644 --- a/nerve/db/tasks.py +++ b/nerve/db/tasks.py @@ -5,6 +5,8 @@ import re from datetime import datetime, timezone +from nerve.identity import UNATTRIBUTED, Attribution + # Spacing between task ranks (``tasks.position``). Wide enough that a # midpoint insert has ~50 levels of headroom before float precision runs @@ -57,9 +59,9 @@ class TaskStore: async def upsert_task( self, task_id: str, - file_path: str, + file_path: str | _Keep, title: str, - status: str = "pending", + status: str | _Keep = "pending", source: str | None | _Keep = KEEP, source_url: str | None | _Keep = KEEP, deadline: str | None | _Keep = KEEP, @@ -69,11 +71,17 @@ async def upsert_task( actor: str = "system", *, expect_revision: int | None = None, + attribution: Attribution = UNATTRIBUTED, ) -> bool: """Insert or update a task row and its FTS entry. ``file_path``, ``title``, ``status`` and ``content`` are a full replace. Every caller rebuilds them from the markdown file. + ``file_path`` and ``status`` also take :data:`KEEP`, which keeps the + stored value, read in the same transaction as the write. A caller that + edits only the content passes it, so that a status change that lands + between its read and this write is kept and is not recorded again. + :data:`KEEP` needs an existing row. ``source``, ``source_url``, ``deadline``, ``tags`` and ``position`` are preserve-on-omit. An omitted column keeps its stored value. To @@ -95,6 +103,9 @@ async def upsert_task( ``revision`` still equals it, and otherwise writes nothing and returns ``False``. Without it the call inserts or updates unconditionally and returns ``True``. + + ``actor`` and ``attribution`` go to the status event that a status + change records; see :meth:`_record_status_event`. """ now = datetime.now(timezone.utc).isoformat() async with self._atomic(): @@ -102,6 +113,10 @@ async def upsert_task( # write. The statement below stays a plain full replace, and the # FTS text sees the tags that actually land in the row. stored = await self._stored_task_columns(task_id) + if stored is None and (file_path is KEEP or status is KEEP): + raise ValueError(f"task {task_id!r} has no row to keep values from") + file_path = _resolve_kept(file_path, stored, "file_path", None) + status = _resolve_kept(status, stored, "status", None) source = _resolve_kept(source, stored, "source", None) source_url = _resolve_kept(source_url, stored, "source_url", None) deadline = _resolve_kept(deadline, stored, "deadline", None) @@ -139,7 +154,9 @@ async def upsert_task( await cursor.close() if expect_revision is not None and (rowcount or 0) != 1: return False - await self._record_status_event(task_id, previous_status, status, actor) + await self._record_status_event( + task_id, previous_status, status, actor, attribution, + ) await self._sync_task_fts(task_id, title, content, tags) return True @@ -151,6 +168,7 @@ async def transition_task( expect: tuple[str, ...] = (), expect_revision: int | None = None, actor: str = "system", + attribution: Attribution = UNATTRIBUTED, content: str | None = None, **fields, ) -> bool: @@ -206,7 +224,9 @@ async def transition_task( await cursor.close() if not applied: return False - await self._record_status_event(task_id, row["status"], to_status, actor) + await self._record_status_event( + task_id, row["status"], to_status, actor, attribution, + ) if content is not None: await self._sync_task_fts( task_id, fields.get("title", row["title"]), content, @@ -241,6 +261,7 @@ async def _record_status_event( from_status: str | None, to_status: str, actor: str, + attribution: Attribution = UNATTRIBUTED, ) -> None: """Append a transition to ``task_events``, if it is one. @@ -254,14 +275,20 @@ async def _record_status_event( is recorded. The single case where reindex *does* change a status — resetting an orphaned row to match its directory — is a real correction and worth a row. + + ``actor`` is the free-text origin of the change (``web``, a session + ID, ``system``). ``attribution`` names the actor that made the change + and the principal it was made for, as ``actor_refs`` IDs. """ if from_status == to_status: return await self.db.execute( - "INSERT INTO task_events (task_id, from_status, to_status, actor, created_at) " - "VALUES (?, ?, ?, ?, ?)", + "INSERT INTO task_events (task_id, from_status, to_status, actor, " + "actor_id, caused_by_actor_id, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", ( task_id, from_status, to_status, actor or "system", + attribution.actor_id, attribution.caused_by_actor_id, datetime.now(timezone.utc).isoformat(), ), ) @@ -325,7 +352,7 @@ async def _lane_top_position(self, status: str) -> float: async def _stored_task_columns(self, task_id: str) -> dict | None: """The columns an upsert can preserve, or ``None`` for a new task.""" async with self.db.execute( - "SELECT status, position, source, source_url, deadline, tags " + "SELECT status, file_path, position, source, source_url, deadline, tags " "FROM tasks WHERE id = ?", (task_id,), ) as cursor: @@ -425,6 +452,7 @@ async def move_task( before_id: str | None = None, after_id: str | None = None, actor: str = "system", + attribution: Attribution = UNATTRIBUTED, ) -> dict | None: """Place a task at an explicit slot in a lane; return the new row. @@ -466,7 +494,7 @@ async def move_task( # A pure reorder leaves status alone, and _record_status_event # drops the no-op — so dragging within a lane doesn't pollute # the history with rows that say nothing changed. - await self._record_status_event(task_id, row[0], lane, actor) + await self._record_status_event(task_id, row[0], lane, actor, attribution) async with self.db.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,), ) as cursor: @@ -561,9 +589,10 @@ async def count_tasks( async def update_task_status( self, task_id: str, status: str, actor: str = "system", + *, attribution: Attribution = UNATTRIBUTED, ) -> None: """An unconditional :meth:`transition_task`; a missing row is a no-op.""" - await self.transition_task(task_id, status, actor=actor) + await self.transition_task(task_id, status, actor=actor, attribution=attribution) async def update_task_tags( self, task_id: str, tags: str, *, expect_revision: int | None = None, diff --git a/nerve/gateway/routes/_deps.py b/nerve/gateway/routes/_deps.py index 42ecca5f4..8887f0569 100644 --- a/nerve/gateway/routes/_deps.py +++ b/nerve/gateway/routes/_deps.py @@ -14,6 +14,7 @@ from nerve.agent.tools import ToolContext, ToolRegistry from nerve.db import Database from nerve.external_agents.sync_service import SyncService + from nerve.identity import Actor from nerve.notifications.service import NotificationService @@ -71,7 +72,9 @@ def get_tool_registry() -> "ToolRegistry": return get_deps().engine.registry -def build_route_tool_context(session_id: str = "system") -> "ToolContext": +def build_route_tool_context( + session_id: str = "system", *, actor: "Actor | None" = None, +) -> "ToolContext": """Construct a :class:`ToolContext` for an HTTP-route-initiated tool call. Routes don't belong to an LLM session, so we use a stable @@ -79,6 +82,10 @@ def build_route_tool_context(session_id: str = "system") -> "ToolContext": reads ``ctx.session_id``. Notification-style tools are *not* invoked from routes today, but if they ever are they'll attribute to that well-known sentinel. + + ``actor`` is the request's actor. A person who calls a route makes the + change for themselves, so it is both the actor and the cause of the + handler's writes. """ from nerve.agent.tools import ToolContext # local import to avoid cycle @@ -94,4 +101,6 @@ def build_route_tool_context(session_id: str = "system") -> "ToolContext": skill_manager=engine._skill_manager, engine=engine, notification_service=deps.notification_service, + actor=actor, + cause=actor, ) diff --git a/nerve/gateway/routes/actors.py b/nerve/gateway/routes/actors.py index a6a07a691..9ef0bc205 100644 --- a/nerve/gateway/routes/actors.py +++ b/nerve/gateway/routes/actors.py @@ -1,4 +1,8 @@ -"""Bulk lookup for the actor ids stored on sessions and messages. +"""Bulk lookup for the actor ids stored in attribution columns. + +The attribution columns are listed in :mod:`nerve.db.actor_columns`. The +endpoint returns every ``actor_refs`` row, so it covers each ID that one of +those columns holds. The response is built field by field from ``actor_refs`` plus the login name of the account behind a human actor, so credential and access state cannot leak diff --git a/nerve/gateway/routes/plans.py b/nerve/gateway/routes/plans.py index cb45dc3f7..0cfcaf383 100644 --- a/nerve/gateway/routes/plans.py +++ b/nerve/gateway/routes/plans.py @@ -111,6 +111,7 @@ async def revise_plan(plan_id: str, req: PlanReviseRequest, actor: Actor = Depen engine=deps.engine, plan_id=plan_id, feedback=req.feedback, + cause=actor, ) except PlanNotFound: raise HTTPException(status_code=404, detail="Plan not found") @@ -231,7 +232,7 @@ async def _run_impl(): # Nerve assembles the prompt from the task file and the # approved plan, so the turn is the instance's own work even # though a person's approval started it. The session above - # carries who approved. + # carries who approved, and its cause is the approver. actor=deps.db.system_actor, ) except Exception: diff --git a/nerve/gateway/routes/tasks.py b/nerve/gateway/routes/tasks.py index eba5a9fc1..f6780f380 100644 --- a/nerve/gateway/routes/tasks.py +++ b/nerve/gateway/routes/tasks.py @@ -10,16 +10,23 @@ from nerve.agent.streaming import emit_task_event from nerve.config import ensure_path_not_tracked_config, get_config from nerve.db.task_statuses import STATUS_NAME_RE, TERMINAL_STATUS, normalize_color +from nerve.db.tasks import KEEP from nerve.gateway.auth import require_auth from nerve.gateway.routes._deps import ( build_route_tool_context, get_deps, get_tool_registry, ) +from nerve.identity import Actor, attribution_for from nerve.tasks.files import write_task_file +from nerve.tasks.locks import task_mutation router = APIRouter(dependencies=[Depends(require_auth)]) +# ``task_events.actor`` for a change made through these routes: the origin of +# the change. The person who made it is in ``task_events.actor_id``. +_ROUTE_ORIGIN = "web" + class TaskCreateRequest(BaseModel): title: str @@ -203,12 +210,12 @@ async def list_task_tags(include_done: bool = False): @router.post("/api/tasks") -async def create_task(req: TaskCreateRequest): +async def create_task(req: TaskCreateRequest, actor: Actor = Depends(require_auth)): # Route into the unified handler surface via the live registry — no # behavior duplication between the REST and MCP paths. result = await get_tool_registry().invoke( "task_create", - build_route_tool_context(), + build_route_tool_context(_ROUTE_ORIGIN, actor=actor), { "title": req.title, "content": req.content, @@ -250,8 +257,17 @@ async def create_task(req: TaskCreateRequest): @router.post("/api/tasks/{task_id}/move") -async def move_task(task_id: str, req: TaskMoveRequest): +async def move_task( + task_id: str, req: TaskMoveRequest, actor: Actor = Depends(require_auth), +): """Reorder a task within a lane, or move it to another one.""" + # One lock for the status change and the rank: a change that lands + # between them could move the file while the rank write keeps a status. + async with task_mutation(task_id): + return await _move_task(task_id, req, actor) + + +async def _move_task(task_id: str, req: TaskMoveRequest, actor: Actor) -> dict: deps = get_deps() task = await deps.db.get_task(task_id) if not task: @@ -271,7 +287,7 @@ async def move_task(task_id: str, req: TaskMoveRequest): # "web" rather than the default "system" sentinel: this string # lands in task_events.actor, and the point of that column is # telling a person dragging a card from the agent moving it. - build_route_tool_context("web"), + build_route_tool_context(_ROUTE_ORIGIN, actor=actor), {"task_id": task_id, "status": target}, ) if result.is_error: @@ -279,7 +295,7 @@ async def move_task(task_id: str, req: TaskMoveRequest): moved = await deps.db.move_task( task_id, status=target, before_id=req.before_id, after_id=req.after_id, - actor="web", + actor=_ROUTE_ORIGIN, attribution=attribution_for(actor), ) if not moved: raise HTTPException(status_code=404, detail="Task not found") @@ -312,7 +328,17 @@ async def get_task(task_id: str): @router.patch("/api/tasks/{task_id}") -async def update_task(task_id: str, req: TaskUpdateRequest): +async def update_task( + task_id: str, req: TaskUpdateRequest, actor: Actor = Depends(require_auth), +): + # One lock for the content save and the field and status changes, from + # the first read of the row: the file this request writes is the file + # that the row names when the request ends. + async with task_mutation(task_id): + return await _update_task(task_id, req, actor) + + +async def _update_task(task_id: str, req: TaskUpdateRequest, actor: Actor) -> dict: deps = get_deps() task = await deps.db.get_task(task_id) if not task: @@ -346,12 +372,18 @@ async def update_task(task_id: str, req: TaskUpdateRequest): edits["deadline"] = fields["deadline"] if fields.get("tags"): edits["tags"] = tags_to_string(parse_tags_string(fields["tags"])) + # KEEP, not ``req.status`` or the status read above: a status + # change goes through task_update below, which moves the file + # into or out of done/ and records the transition once, for this + # person. A move by somebody else after the read is kept. await deps.db.upsert_task( task_id=task_id, - file_path=task["file_path"], + file_path=KEEP, title=new_title, - status=req.status or task["status"], + status=KEEP, content=req.content, + actor=_ROUTE_ORIGIN, + attribution=attribution_for(actor), **edits, ) @@ -377,7 +409,7 @@ async def update_task(task_id: str, req: TaskUpdateRequest): if len(payload) > 1: result = await get_tool_registry().invoke( - "task_update", build_route_tool_context("web"), payload, + "task_update", build_route_tool_context(_ROUTE_ORIGIN, actor=actor), payload, ) # Previously swallowed: an unknown status returned {"updated": true} # while nothing had changed. Surface the handler's own message. @@ -402,7 +434,9 @@ async def list_task_statuses(): @router.post("/api/task-statuses") -async def create_task_status(req: TaskStatusCreateRequest): +async def create_task_status( + req: TaskStatusCreateRequest, actor: Actor = Depends(require_auth), +): deps = get_deps() name = (req.name or "").strip().lower() if not STATUS_NAME_RE.match(name): @@ -419,6 +453,7 @@ async def create_task_status(req: TaskStatusCreateRequest): created = await deps.db.create_task_status( name=name, label=label, color=color, description=(req.description or "").strip(), + attribution=attribution_for(actor), ) return created @@ -430,17 +465,24 @@ class TaskStatusReorderRequest(BaseModel): @router.post("/api/task-statuses/reorder") -async def reorder_task_statuses(req: TaskStatusReorderRequest): +async def reorder_task_statuses( + req: TaskStatusReorderRequest, actor: Actor = Depends(require_auth), +): """Set board column order in one write. Declared above /{name} so "reorder" isn't captured as a status name. """ deps = get_deps() - return {"statuses": await deps.db.reorder_task_statuses(req.names)} + statuses = await deps.db.reorder_task_statuses( + req.names, attribution=attribution_for(actor), + ) + return {"statuses": statuses} @router.patch("/api/task-statuses/{name}") -async def update_task_status(name: str, req: TaskStatusUpdateRequest): +async def update_task_status( + name: str, req: TaskStatusUpdateRequest, actor: Actor = Depends(require_auth), +): deps = get_deps() existing = await deps.db.get_task_status_def(name) if not existing: @@ -452,12 +494,13 @@ async def update_task_status(name: str, req: TaskStatusUpdateRequest): color=req.color if req.color is not None else None, description=req.description if req.description is not None else None, sort_order=req.sort_order, + attribution=attribution_for(actor), ) return await deps.db.get_task_status_def(name) @router.delete("/api/task-statuses/{name}") -async def delete_task_status(name: str): +async def delete_task_status(name: str, actor: Actor = Depends(require_auth)): deps = get_deps() existing = await deps.db.get_task_status_def(name) if not existing: @@ -474,5 +517,5 @@ async def delete_task_status(name: str): detail=f"{in_use} task(s) use the '{name}' status. " "Reassign them before deleting it.", ) - await deps.db.delete_task_status_def(name) + await deps.db.delete_task_status_def(name, attribution=attribution_for(actor)) return {"name": name, "deleted": True} diff --git a/nerve/identity.py b/nerve/identity.py index 118fb2a36..828f8bf59 100644 --- a/nerve/identity.py +++ b/nerve/identity.py @@ -17,12 +17,21 @@ taken when the actor was resolved. Renaming an account changes what later requests carry and rewrites nothing stored before. Names are never identity or authorization keys. + +A write records two IDs. The *actor* made the change: a person through the UI, +REST or the browser WebSocket, or the system actor for a tool call, a cron job +or any other internal process. The *cause* is the principal on whose behalf +the work happens: the person for a change that the person made, and for agent +work the cause of its session (``sessions.caused_by_actor_id``), which is the +person who started the session or the system actor for a session that the +instance started for itself. ``None`` in either position means unknown. +:func:`attribution_for` builds the pair for one write. """ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, TypeAlias if TYPE_CHECKING: # pragma: no cover - typing only, never imported at runtime from nerve.db.accounts import AccountStore @@ -69,6 +78,72 @@ def is_human(self) -> bool: return self.kind == ACTOR_KIND_HUMAN +# An actor, or the ``actor_refs`` ID of one. A cause that comes from storage +# (a session's stored cause, for example) is an ID; a cause that comes from a +# request is the resolved :class:`Actor`. +ActorOrId: TypeAlias = "Actor | str" + + +class CauseFromActor: + """Type of the :data:`CAUSE_FROM_ACTOR` sentinel.""" + + __slots__ = () + + def __repr__(self) -> str: + return "CAUSE_FROM_ACTOR" + + +# Default for a cause argument: the cause is the actor. A person acts for +# themselves, the system actor acts for itself, and no actor gives no cause. +# A separate sentinel is necessary because ``None`` is a real cause: unknown. +CAUSE_FROM_ACTOR = CauseFromActor() + +CauseArg: TypeAlias = "ActorOrId | None | CauseFromActor" + + +def actor_id_of(ref: ActorOrId | None) -> str | None: + """The actor ID that ``ref`` names, or ``None`` for no actor.""" + if ref is None: + return None + if isinstance(ref, Actor): + return ref.actor_id + if isinstance(ref, str) and ref: + return ref + raise ValueError(f"an actor reference must be an Actor or an actor ID, got {ref!r}") + + +def resolve_cause(actor: ActorOrId | None, cause: CauseArg = CAUSE_FROM_ACTOR) -> str | None: + """The cause actor ID for one write, or for a session that is created.""" + if isinstance(cause, CauseFromActor): + return actor_id_of(actor) + return actor_id_of(cause) + + +@dataclass(frozen=True, slots=True) +class Attribution: + """The ``(actor_id, caused_by_actor_id)`` pair that one write records. + + Every store method that records attribution takes one of these, so that + all writes get the two values from the same rule. The store maps the two + fields to its own column names. + """ + + actor_id: str | None = None + caused_by_actor_id: str | None = None + + +# Nothing is known about who made the change. +UNATTRIBUTED = Attribution() + + +def attribution_for(actor: ActorOrId | None, cause: CauseArg = CAUSE_FROM_ACTOR) -> Attribution: + """Build the pair for a write by ``actor`` on behalf of ``cause``.""" + return Attribution( + actor_id=actor_id_of(actor), + caused_by_actor_id=resolve_cause(actor, cause), + ) + + def _actor_for_account_row(account: dict) -> Actor: """Resolve an account row, checking disablement on every request.""" if not account.get("enabled"): diff --git a/nerve/mcp_server/http.py b/nerve/mcp_server/http.py index 0adbd91f2..bc862de7d 100644 --- a/nerve/mcp_server/http.py +++ b/nerve/mcp_server/http.py @@ -188,7 +188,6 @@ async def _resolve(rctx: ServerRequestContext | None = None) -> ToolContext: session_id, runtime_metadata = _bound_identity_from_request( engine.config, rctx ) - if session_id is None: client_name, mcp_session_id, _ = _resolve_client_info(rctx) @@ -206,6 +205,9 @@ async def _resolve(rctx: ServerRequestContext | None = None) -> ToolContext: "mcp_session_id": mcp_session_id, } + # The call is the agent's change for the session's stored cause. A + # satellite session stores none: its agent runs outside this instance. + session = await engine.db.get_session(session_id) return ToolContext( session_id=session_id, workspace=engine.config.workspace, @@ -217,6 +219,8 @@ async def _resolve(rctx: ServerRequestContext | None = None) -> ToolContext: engine=engine, notification_service=engine.notification_service, runtime_metadata=runtime_metadata, + actor=engine.db.system_actor, + cause=(session or {}).get("caused_by_actor_id"), ) return _resolve diff --git a/nerve/mcp_server/session.py b/nerve/mcp_server/session.py index fbecf9405..0b2cfbd27 100644 --- a/nerve/mcp_server/session.py +++ b/nerve/mcp_server/session.py @@ -152,6 +152,8 @@ async def resolve( metadata=metadata, status="active", actor=self.db.system_actor, + # The agent behind it runs outside this instance. + cause=None, ) logger.info( "Created Codex satellite session %s via MCP (mcp=%s)", @@ -183,6 +185,8 @@ async def resolve( metadata=metadata, status="active", actor=self.db.system_actor, + # The agent behind it runs outside this instance. + cause=None, ) logger.info( "Created satellite session %s (client=%s, mcp=%s)", diff --git a/nerve/sources/codex_threads/ingester.py b/nerve/sources/codex_threads/ingester.py index 0823f4662..17d2fdad4 100644 --- a/nerve/sources/codex_threads/ingester.py +++ b/nerve/sources/codex_threads/ingester.py @@ -203,8 +203,9 @@ async def _ensure_session(self, event: ThreadEvent) -> None: backend="codex", cwd=meta.cwd, # The sync created this row, not a person — the Codex thread it - # mirrors was started outside Nerve entirely. + # mirrors was started outside Nerve entirely, by nobody known. actor=self.db.system_actor, + cause=None, ) await self.db.bind_native_thread("codex", event.thread_id, session_id) self.stats["threads_in_scope"] += 1 diff --git a/nerve/tasks/locks.py b/nerve/tasks/locks.py new file mode 100644 index 000000000..b64a90214 --- /dev/null +++ b/nerve/tasks/locks.py @@ -0,0 +1,45 @@ +"""One lock per task for the changes that write a task file and its row. + +A task change reads the row, writes or moves the markdown file, and writes +the row. Two changes of one task that interleave can strand an edit: one of +them writes the path it read after the other has moved the file into or out +of ``done/``, so the edit goes to a file that the row no longer names. Every +such change runs inside :func:`task_mutation`, so the changes of one task +run one at a time. Changes of different tasks do not wait for each other. + +The lock is re-entrant for the asyncio task that holds it: a handler that +calls another handler for the same task (``task_update`` calls +``task_done``) does not wait for itself. Another asyncio task waits. The +locks live in this process; Nerve runs one gateway process. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import weakref +from collections.abc import AsyncIterator + +# A lock exists while some caller holds or waits for it. +_locks: "weakref.WeakValueDictionary[str, asyncio.Lock]" = weakref.WeakValueDictionary() +# The asyncio task that holds each task's lock. +_owners: dict[str, asyncio.Task] = {} + + +@contextlib.asynccontextmanager +async def task_mutation(task_id: str) -> AsyncIterator[None]: + """Hold the lock of ``task_id`` for one file-plus-row change.""" + me = asyncio.current_task() + if me is not None and _owners.get(task_id) is me: + yield + return + lock = _locks.get(task_id) + if lock is None: + lock = asyncio.Lock() + _locks[task_id] = lock + async with lock: + _owners[task_id] = me + try: + yield + finally: + _owners.pop(task_id, None) diff --git a/nerve/tasks/manager.py b/nerve/tasks/manager.py index 4893af0c5..b1350e887 100644 --- a/nerve/tasks/manager.py +++ b/nerve/tasks/manager.py @@ -9,7 +9,9 @@ from nerve.config import ensure_path_not_tracked_config from nerve.db import Database +from nerve.identity import Attribution, attribution_for from nerve.tasks.files import move_task_file +from nerve.tasks.locks import task_mutation from nerve.tasks.models import ( Task, TaskStatus, @@ -33,6 +35,10 @@ def __init__(self, workspace: Path, db: Database): self.active_dir.mkdir(parents=True, exist_ok=True) self.done_dir.mkdir(parents=True, exist_ok=True) + def _own_attribution(self) -> Attribution: + """This manager's writes are Nerve's own work: system for system.""" + return attribution_for(self.db.system_actor) + async def reindex(self) -> int: """Scan task files and rebuild the SQLite + FTS index. @@ -94,6 +100,7 @@ async def reindex(self) -> int: title=title, status=status, content=content, + attribution=self._own_attribution(), **edits, ) count += 1 @@ -123,6 +130,11 @@ async def get_task(self, task_id: str) -> Task | None: return task async def mark_done(self, task_id: str, expect_revision: int | None = None) -> bool: + """Mark a task as done and move its file, under the task's lock.""" + async with task_mutation(task_id): + return await self._mark_done(task_id, expect_revision) + + async def _mark_done(self, task_id: str, expect_revision: int | None) -> bool: """Mark a task as done and move its file. Returns ``False`` when the task is missing, or when ``expect_revision`` @@ -155,10 +167,12 @@ async def mark_done(self, task_id: str, expect_revision: int | None = None) -> b await asyncio.to_thread(move) return await self.db.transition_task( task_id, "done", file_path=rel_path, content=content, + attribution=self._own_attribution(), ) if not await self.db.transition_task( task_id, "done", expect_revision=expect_revision, file_path=rel_path, content=content, + attribution=self._own_attribution(), ): return False await asyncio.to_thread(move) @@ -166,6 +180,7 @@ async def mark_done(self, task_id: str, expect_revision: int | None = None) -> b return await self.db.transition_task( task_id, "done", expect_revision=expect_revision, + attribution=self._own_attribution(), ) async def get_overdue_tasks(self) -> list[Task]: diff --git a/tests/test_attribution.py b/tests/test_attribution.py index dee81c07a..35b61c928 100644 --- a/tests/test_attribution.py +++ b/tests/test_attribution.py @@ -76,6 +76,9 @@ def app(self) -> FastAPI: async def creator_of(self, session_id: str) -> str | None: return (await self.db.get_session(session_id))["created_by_actor_id"] + async def cause_of(self, session_id: str) -> str | None: + return (await self.db.get_session(session_id))["caused_by_actor_id"] + async def messages_in(self, session_id: str) -> list[dict]: return await self.db.get_messages(session_id) @@ -325,6 +328,7 @@ async def test_websocket_created_session_belongs_to_the_connection( await _ws_endpoint()(socket) switched = next(msg for msg in socket.sent if msg["type"] == "session_switched") assert await install.creator_of(switched["session_id"]) == install.bob_actor + assert await install.cause_of(switched["session_id"]) == install.bob_actor async def _wait_for_user_messages(install, session_id: str, count: int) -> None: @@ -345,6 +349,7 @@ async def test_autonomous_cron_uses_the_system_actor(install, monkeypatch): if row["source"] == "cron" ) assert session["created_by_actor_id"] == install.system_actor_id + assert session["caused_by_actor_id"] == install.system_actor_id assert [(row["role"], row["actor_id"]) for row in await install.messages_in(session["id"])] == [ ("user", install.system_actor_id), ("assistant", None), @@ -379,6 +384,7 @@ async def test_unidentified_channel_human_stays_unattributed( )) session_id = await install.engine.sessions.get_last_session(key) assert await install.creator_of(session_id) is None + assert await install.cause_of(session_id) is None assert await install.said_in(session_id) == [("hello from a chat app", None)] @@ -440,6 +446,8 @@ async def test_codex_sync_session_is_system_but_imported_human_is_unknown(instal session_id = "codex:thread-aaa" assert await install.creator_of(session_id) == install.system_actor_id + # Nobody known started the thread that the session imports. + assert await install.cause_of(session_id) is None assert [(row["role"], row["actor_id"]) for row in await install.messages_in(session_id)] == [ ("user", None), ("assistant", None), diff --git a/tests/test_cause_attribution.py b/tests/test_cause_attribution.py new file mode 100644 index 000000000..a16fe4d70 --- /dev/null +++ b/tests/test_cause_attribution.py @@ -0,0 +1,1116 @@ +"""Actor and cause attribution: the session cause and task attribution. + +A write records who made it (``actor_id``) and on whose behalf +(``caused_by_actor_id``). A person who changes something directly is both. A +tool call by the agent is the system actor's change, made for the cause of its +session (``sessions.caused_by_actor_id``), which is fixed when the session is +created. These tests drive the real routes, engine, tool registry and +database; only the model runtime is replaced, by a client factory that makes +tool calls the way the agent would inside a turn. +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import threading +from types import SimpleNamespace +from unittest.mock import patch + +import httpx +import pytest +import pytest_asyncio +from fastapi import FastAPI + +from nerve.agent.backends import TurnCompleted +from nerve.agent.engine import AgentEngine +from nerve.agent.tools.handlers.plans import plan_approve_handler +from nerve.agent.tools.registry import ToolContext +from nerve.config import NerveConfig, set_config +from nerve.db import Database +from nerve.db.actor_columns import ACTOR_COLUMNS +from nerve.gateway.auth import ( + ACTOR_CONTEXT_HEADER, + AUTH_MODE_ENV, + AUTH_MODE_EXTERNAL, + create_session_token, + hash_password, + pin_auth_mode, + pin_jwt_secret, +) +from nerve.gateway.routes import init_deps, register_all_routes +from nerve.identity import ( + CAUSE_FROM_ACTOR, + UNATTRIBUTED, + Actor, + Attribution, + actor_id_of, + attribution_for, + resolve_cause, +) +from nerve.mcp_server.http import build_ctx_resolver +from nerve.mcp_server.session import SatelliteSessionResolver +from nerve.migrate import bootstrap_identity + +_SECRET = "test-secret-for-cause-attribution-pad-32b" +# One hash for both people: the tests sign in with tokens, and a hash per +# fixture is the slowest part of the setup. +_PASSWORD_HASH = hash_password("correct-horse-battery-staple") + +# External-mode principals (synthetic). +ALICE_PRINCIPAL = "6f1c2a8e-3b4d-4e5f-8a9b-0c1d2e3f4a5b" +BOB_PRINCIPAL = "0a9b8c7d-6e5f-4a3b-9c2d-1e0f9a8b7c6d" + +ALICE = Actor(actor_id="00000000-0000-4000-8000-0000000a11ce", kind="human") +SYSTEM = Actor(actor_id="00000000-0000-4000-8000-00000000515e", kind="system") + + +# --------------------------------------------------------------------------- # +# The helper and the tool context # +# --------------------------------------------------------------------------- # + + +class TestAttributionHelper: + def test_the_cause_defaults_to_the_actor(self): + assert attribution_for(ALICE) == Attribution(ALICE.actor_id, ALICE.actor_id) + assert attribution_for(SYSTEM) == Attribution(SYSTEM.actor_id, SYSTEM.actor_id) + assert attribution_for(None) == UNATTRIBUTED + + def test_an_explicit_cause_wins_and_none_stays_unknown(self): + assert attribution_for(SYSTEM, ALICE) == Attribution( + SYSTEM.actor_id, ALICE.actor_id, + ) + assert attribution_for(SYSTEM, None) == Attribution(SYSTEM.actor_id, None) + + def test_a_cause_can_be_a_stored_id(self): + assert resolve_cause(SYSTEM, ALICE.actor_id) == ALICE.actor_id + assert resolve_cause(SYSTEM, CAUSE_FROM_ACTOR) == SYSTEM.actor_id + + @pytest.mark.parametrize("bad", ["", 42, object()]) + def test_a_reference_that_names_no_actor_is_refused(self, bad): + with pytest.raises(ValueError): + actor_id_of(bad) + + +class TestToolContextCause: + def test_a_fixed_cause(self): + ctx = ToolContext(session_id="s", actor=SYSTEM, cause=ALICE) + assert ctx.cause_id == ALICE.actor_id + assert ctx.attribution == Attribution(SYSTEM.actor_id, ALICE.actor_id) + + def test_a_callable_cause_is_read_at_the_time_of_the_call(self): + current = {"cause": ALICE.actor_id} + ctx = ToolContext(session_id="s", actor=SYSTEM, cause=lambda: current["cause"]) + assert ctx.cause_id == ALICE.actor_id + current["cause"] = None + assert ctx.attribution == Attribution(SYSTEM.actor_id, None) + + def test_a_context_without_identity_is_unattributed(self): + assert ToolContext(session_id="s").attribution == UNATTRIBUTED + + +# --------------------------------------------------------------------------- # +# Local-mode install with two people # +# --------------------------------------------------------------------------- # + + +class _FakeRuntime: + """A model runtime client that makes tool calls the way the agent does. + + The engine uses it through the client API (``start_turn``, + ``receive_turn``, the idle-event calls). Each turn calls ``task_create`` + through the session's own tool context, as the Claude CLI calls the + in-process MCP server, and then completes. ``before_tool_call`` runs + first when it is set. + """ + + native_session_id = None + + def __init__(self, install: "_Install", session_id: str): + self.install = install + self.session_id = session_id + self.before_tool_call = None + + def try_receive_idle_events(self): + return None + + async def start_turn(self, turn_input) -> None: + return None + + async def disconnect(self) -> None: + return None + + async def receive_turn(self): + if self.before_tool_call is not None: + await self.before_tool_call() + await self.install.agent_tool_call(self.session_id) + yield TurnCompleted() + + +class _Install: + def __init__(self, db: Database, engine: AgentEngine, config: NerveConfig): + self.db = db + self.engine = engine + self.config = config + self.people: dict[str, dict] = {} + # One record per tool call of the fake runtime. + self.turns: list[dict] = [] + self.runtimes: dict[str, _FakeRuntime] = {} + + @property + def system_actor_id(self) -> str: + return self.db.system_actor_id + + def actor_id(self, name: str) -> str: + return self.people[name]["actor_id"] + + def headers(self, name: str) -> dict: + token = create_session_token(_SECRET, self.people[name]["id"]) + return {"Authorization": f"Bearer {token}"} + + def app(self) -> FastAPI: + app = FastAPI() + app.include_router(register_all_routes()) + return app + + async def events(self, task_id: str) -> list[tuple]: + return [ + (row["from_status"], row["to_status"], row["actor"], + row["actor_id"], row["caused_by_actor_id"]) + for row in await self.db.list_task_events(task_id) + ] + + async def user_rows(self, session_id: str) -> list[tuple]: + return [ + (row["content"], row["actor_id"]) + for row in await self.db.get_messages(session_id) + if row["role"] == "user" + ] + + async def agent_tool_call(self, session_id: str) -> None: + """The agent calls ``task_create`` in a turn of the session.""" + engine = self.engine + cause = engine.session_cause(session_id) + result = await engine.registry.invoke( + "task_create", engine._build_tool_context(session_id), + {"title": f"Made in {session_id} #{len(self.turns)}", + "confirm_duplicate": True}, + ) + self.turns.append({ + "session_id": session_id, + "cause": cause, + "task_id": (result.structured or {}).get("task_id"), + }) + + def use_fake_runtime(self, monkeypatch) -> None: + """Serve every session's turns from a :class:`_FakeRuntime`.""" + + async def _client(session_id, source, model, **kwargs): + if session_id not in self.runtimes: + self.runtimes[session_id] = _FakeRuntime(self, session_id) + return self.runtimes[session_id] + + monkeypatch.setattr(self.engine, "_get_or_create_client", _client) + + def runtime(self, session_id: str) -> _FakeRuntime: + if session_id not in self.runtimes: + self.runtimes[session_id] = _FakeRuntime(self, session_id) + return self.runtimes[session_id] + + def turns_in(self, session_id: str) -> list[dict]: + return [turn for turn in self.turns if turn["session_id"] == session_id] + + def turn_in(self, session_id: str) -> dict: + (turn,) = self.turns_in(session_id) + return turn + + async def session_cause_of(self, session_id: str) -> str | None: + return (await self.db.get_session(session_id))["caused_by_actor_id"] + + +@pytest_asyncio.fixture +async def install(tmp_path, open_identity_db, wire_identity_store): + config = NerveConfig.from_dict({ + "workspace": str(tmp_path / "ws"), + "codex": {"home_dir": str(tmp_path / "codex-home")}, + "auth": {"jwt_secret": _SECRET}, + }) + (config.workspace / "memory" / "tasks" / "active").mkdir(parents=True) + set_config(config) + pin_jwt_secret(_SECRET) + database, identity = await open_identity_db(tmp_path / "nerve.db") + wire_identity_store(database) + engine = AgentEngine(config, database) + init_deps(engine, database) + installed = _Install(database, engine, config) + await database.update_account_login( + identity.owner_account_id, username="alice", credential=_PASSWORD_HASH, + ) + installed.people["alice"] = { + "id": identity.owner_account_id, "actor_id": identity.owner_actor_id, + } + bob = await database.create_managed_account( + username="bob", credential=_PASSWORD_HASH, display_name="Bob", + ) + installed.people["bob"] = bob + try: + yield installed + finally: + await database.close() + set_config(NerveConfig()) + + +def _client(app: FastAPI) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://nerve-test", + ) + + +async def _new_session(client, headers) -> str: + return (await client.post("/api/sessions", headers=headers, json={})).json()["id"] + + +async def _chat(client, headers, session_id: str, message: str) -> None: + response = await client.post( + "/api/chat", headers=headers, + json={"session_id": session_id, "message": message}, + ) + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- # +# Schema # +# --------------------------------------------------------------------------- # + + +async def _actor_ref_columns(db: Database) -> set[tuple[str, str]]: + async with db.db.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ) as cursor: + tables = [row[0] async for row in cursor] + found = set() + for table in tables: + async with db.db.execute(f"PRAGMA foreign_key_list({table})") as cursor: + for row in await cursor.fetchall(): + if row["table"] == "actor_refs": + found.add((table, row["from"])) + return found + + +@pytest.mark.asyncio +async def test_the_registry_lists_every_attribution_column(install): + assert len(set(ACTOR_COLUMNS)) == len(ACTOR_COLUMNS) + in_schema = await _actor_ref_columns(install.db) + # accounts.actor_id binds a login to its person; it is not attribution. + assert in_schema - {("accounts", "actor_id")} == set(ACTOR_COLUMNS) + + +@pytest.mark.asyncio +async def test_v053_keeps_old_rows_null_and_is_idempotent(install): + raw = install.db.db + await raw.execute("ALTER TABLE sessions DROP COLUMN caused_by_actor_id") + await raw.execute("ALTER TABLE task_events DROP COLUMN actor_id") + await raw.execute("ALTER TABLE task_events DROP COLUMN caused_by_actor_id") + 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 messages (session_id, role, content, created_at)" + " VALUES ('old', 'user', 'from before', '2020-01-01')" + ) + await raw.execute( + "INSERT INTO task_events (task_id, from_status, to_status, actor, created_at)" + " VALUES ('old-task', NULL, 'pending', 'web', '2020-01-01')" + ) + await raw.commit() + + from nerve.db.migrations.v053_cause_attribution import up + + await up(raw) + await up(raw) + await raw.commit() + assert await install.session_cause_of("old") is None + assert await install.user_rows("old") == [("from before", None)] + assert await install.events("old-task") == [(None, "pending", "web", None, None)] + + +# --------------------------------------------------------------------------- # +# Task changes through REST # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_rest_task_changes_record_each_person(install): + alice, bob = install.actor_id("alice"), install.actor_id("bob") + async with _client(install.app()) as client: + created = await client.post( + "/api/tasks", headers=install.headers("alice"), + json={"title": "Write the report"}, + ) + assert created.status_code == 200 + task_id = created.json()["task"]["id"] + moved = await client.post( + f"/api/tasks/{task_id}/move", headers=install.headers("bob"), + json={"status": "in_progress"}, + ) + assert moved.status_code == 200 + events = ( + await client.get(f"/api/tasks/{task_id}/events", headers=install.headers("bob")) + ).json()["events"] + + assert [ + (e["to_status"], e["actor"], e["actor_id"], e["caused_by_actor_id"]) + for e in events + ] == [ + ("pending", "web", alice, alice), + ("in_progress", "web", bob, bob), + ] + + +@pytest.mark.asyncio +async def test_rest_patch_with_content_records_one_change_for_the_person(install): + alice, bob = install.actor_id("alice"), install.actor_id("bob") + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), + json={"title": "Edit and move"}, + )).json()["task"]["id"] + patched = await client.patch( + f"/api/tasks/{task_id}", headers=install.headers("bob"), + json={"content": "# Edit and move\n\nNew body.\n", "status": "deferred"}, + ) + assert patched.status_code == 200 + finished = await client.patch( + f"/api/tasks/{task_id}", headers=install.headers("alice"), + json={"content": "# Edit and move\n\nBody again.\n", "status": "done"}, + ) + assert finished.status_code == 200 + assert (await install.db.get_task(task_id))["file_path"].startswith( + "memory/tasks/done/", + ) + reopened = await client.patch( + f"/api/tasks/{task_id}", headers=install.headers("bob"), + json={"content": "# Edit and move\n\nOnce more.\n", "status": "pending"}, + ) + assert reopened.status_code == 200 + + assert await install.events(task_id) == [ + (None, "pending", "web", alice, alice), + ("pending", "deferred", "web", bob, bob), + ("deferred", "done", "web", alice, alice), + ("done", "pending", "web", bob, bob), + ] + # Leaving done goes through the reopen path, so the file leaves done/. + task = await install.db.get_task(task_id) + assert task["file_path"].startswith("memory/tasks/active/") + assert (install.config.workspace / task["file_path"]).is_file() + + +def _bob_moves_during_the_file_write(install, monkeypatch, module, task_id: str) -> None: + """Bob moves the task to deferred while ``module`` writes the task file.""" + loop = asyncio.get_running_loop() + real_write = module.write_task_file + + def write_while_bob_moves(path, content): + real_write(path, content) + asyncio.run_coroutine_threadsafe( + install.db.move_task( + task_id, status="deferred", actor="web", + attribution=attribution_for(install.actor_id("bob")), + ), + loop, + ).result(timeout=5) + + monkeypatch.setattr(module, "write_task_file", write_while_bob_moves) + + +async def _assert_bobs_move_stays(install, task_id: str) -> None: + alice, bob = install.actor_id("alice"), install.actor_id("bob") + assert await install.events(task_id) == [ + (None, "pending", "web", alice, alice), + ("pending", "deferred", "web", bob, bob), + ] + assert (await install.db.get_task(task_id))["status"] == "deferred" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("edit", [ + {"title": "Raced, renamed"}, + {"note": "a note"}, + {"deadline": "2026-12-01"}, + {"tags": "alpha"}, +]) +async def test_rest_patch_fields_keep_a_concurrent_move(install, monkeypatch, edit): + from nerve.agent.tools.handlers import tasks as task_handlers + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + _bob_moves_during_the_file_write(install, monkeypatch, task_handlers, task_id) + response = await client.patch( + f"/api/tasks/{task_id}", headers=install.headers("alice"), json=edit, + ) + assert response.status_code == 200 + await _assert_bobs_move_stays(install, task_id) + + +@pytest.mark.asyncio +async def test_task_write_keeps_a_concurrent_move(install, monkeypatch): + from nerve.agent.tools.handlers import tasks as task_handlers + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + ctx = ToolContext( + session_id="alices-turn", workspace=install.config.workspace, + db=install.db, config=install.config, engine=install.engine, + actor=install.db.system_actor, cause=install.actor_id("alice"), + ) + await task_handlers.task_read_handler(ctx, {"task_id": task_id}) + _bob_moves_during_the_file_write(install, monkeypatch, task_handlers, task_id) + result = await task_handlers.task_write_handler( + ctx, {"task_id": task_id, "content": "# Raced\n\nWritten by the agent.\n"}, + ) + assert not result.is_error + await _assert_bobs_move_stays(install, task_id) + + +@pytest.mark.asyncio +async def test_rest_patch_content_keeps_a_concurrent_move(install, monkeypatch): + """A move that lands while PATCH writes the file is not undone. + + Alice saves new content while Bob moves the card. The content save must + not write back the status that it read before Bob's move. + """ + from nerve.gateway.routes import tasks as tasks_route + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + _bob_moves_during_the_file_write(install, monkeypatch, tasks_route, task_id) + response = await client.patch( + f"/api/tasks/{task_id}", headers=install.headers("alice"), + json={"content": "# Raced\n\nNew body.\n"}, + ) + assert response.status_code == 200 + await _assert_bobs_move_stays(install, task_id) + task = await install.db.get_task(task_id) + assert "New body." in (install.config.workspace / task["file_path"]).read_text() + + +def _pause_file_writes(monkeypatch, module) -> tuple[threading.Event, threading.Event]: + """Make ``module``'s task file write wait until the test releases it.""" + entered, release = threading.Event(), threading.Event() + real_write = module.write_task_file + + def paused_write(path, content): + entered.set() + assert release.wait(5), "the test never released the write" + real_write(path, content) + + monkeypatch.setattr(module, "write_task_file", paused_write) + return entered, release + + +async def _move_while_the_edit_writes(client, install, task_id, status, edit, entered, release): + """Start ``edit``, move the task while its file write waits, then finish.""" + editing = asyncio.create_task(edit) + assert await asyncio.to_thread(entered.wait, 5) + moving = asyncio.create_task(client.post( + f"/api/tasks/{task_id}/move", headers=install.headers("bob"), + json={"status": status}, + )) + await asyncio.sleep(0.2) + # The move waits for the edit: both change the file and the row. + assert not moving.done() + release.set() + return await editing, await moving + + +def _task_file(install, lane: str, task_id: str): + return install.config.workspace / "memory" / "tasks" / lane / f"{task_id}.md" + + +@pytest.mark.asyncio +async def test_an_agent_edit_and_a_move_into_done_do_not_lose_the_edit( + install, monkeypatch, +): + from nerve.agent.tools.handlers import tasks as task_handlers + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + ctx = ToolContext( + session_id="alices-turn", workspace=install.config.workspace, + db=install.db, config=install.config, engine=install.engine, + actor=install.db.system_actor, cause=install.actor_id("alice"), + ) + await task_handlers.task_read_handler(ctx, {"task_id": task_id}) + entered, release = _pause_file_writes(monkeypatch, task_handlers) + written, moved = await _move_while_the_edit_writes( + client, install, task_id, "done", + task_handlers.task_write_handler( + ctx, {"task_id": task_id, "content": "# Raced\n\nAgent edit.\n"}, + ), + entered, release, + ) + + assert not written.is_error and moved.status_code == 200 + task = await install.db.get_task(task_id) + assert task["status"] == "done" + assert task["file_path"] == f"memory/tasks/done/{task_id}.md" + assert "Agent edit." in _task_file(install, "done", task_id).read_text() + assert not _task_file(install, "active", task_id).exists() + + +@pytest.mark.asyncio +async def test_a_rest_title_edit_and_a_move_into_done_do_not_lose_the_edit( + install, monkeypatch, +): + from nerve.agent.tools.handlers import tasks as task_handlers + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + entered, release = _pause_file_writes(monkeypatch, task_handlers) + patched, moved = await _move_while_the_edit_writes( + client, install, task_id, "done", + client.patch( + f"/api/tasks/{task_id}", headers=install.headers("alice"), + json={"title": "Raced and renamed"}, + ), + entered, release, + ) + + assert patched.status_code == 200 and moved.status_code == 200 + task = await install.db.get_task(task_id) + assert (task["status"], task["title"]) == ("done", "Raced and renamed") + assert "# Raced and renamed" in _task_file(install, "done", task_id).read_text() + assert not _task_file(install, "active", task_id).exists() + + +@pytest.mark.asyncio +async def test_a_rest_content_edit_and_a_move_out_of_done_do_not_lose_the_edit( + install, monkeypatch, +): + from nerve.gateway.routes import tasks as tasks_route + + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Raced"}, + )).json()["task"]["id"] + await client.post( + f"/api/tasks/{task_id}/move", headers=install.headers("alice"), + json={"status": "done"}, + ) + entered, release = _pause_file_writes(monkeypatch, tasks_route) + patched, moved = await _move_while_the_edit_writes( + client, install, task_id, "pending", + client.patch( + f"/api/tasks/{task_id}", headers=install.headers("alice"), + json={"content": "# Raced\n\nEdited while done.\n"}, + ), + entered, release, + ) + + assert patched.status_code == 200 and moved.status_code == 200 + task = await install.db.get_task(task_id) + assert task["status"] == "pending" + assert task["file_path"] == f"memory/tasks/active/{task_id}.md" + assert "Edited while done." in _task_file(install, "active", task_id).read_text() + assert not _task_file(install, "done", task_id).exists() + + +@pytest.mark.asyncio +async def test_task_status_definition_changes_record_actor_and_cause(install): + from nerve.agent.tools.handlers.tasks import task_status_create_handler + + alice, bob = install.actor_id("alice"), install.actor_id("bob") + async with _client(install.app()) as client: + created = await client.post( + "/api/task-statuses", headers=install.headers("bob"), + json={"name": "review", "label": "Review", "color": "#123456"}, + ) + assert created.status_code == 200 + updated = await client.patch( + "/api/task-statuses/review", headers=install.headers("alice"), + json={"label": "In review"}, + ) + assert updated.status_code == 200 + reordered = await client.post( + "/api/task-statuses/reorder", headers=install.headers("bob"), + json={"names": ["review", "pending"]}, + ) + assert reordered.status_code == 200 + deleted = await client.delete( + "/api/task-statuses/review", headers=install.headers("alice"), + ) + assert deleted.status_code == 200 + + # The agent adds a status in a turn that Alice started. + ctx = ToolContext( + session_id="alices-turn", db=install.db, + actor=install.db.system_actor, cause=alice, + ) + await task_status_create_handler(ctx, {"name": "blocked"}) + + events = await install.db.list_task_status_events() + assert [ + (e["change"], e["status_name"], e["actor_id"], e["caused_by_actor_id"]) + for e in events + ] == [ + ("created", "review", bob, bob), + ("updated", "review", alice, alice), + ("reordered", None, bob, bob), + ("deleted", "review", alice, alice), + ("created", "blocked", install.system_actor_id, alice), + ] + assert json.loads(events[1]["detail"]) == {"label": "In review"} + assert json.loads(events[2]["detail"])["order"][:2] == ["review", "pending"] + + +# --------------------------------------------------------------------------- # +# The session cause: stored once, used by every agent action in the session # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_a_person_creates_and_forks_sessions_for_themselves(install): + alice, bob = install.actor_id("alice"), install.actor_id("bob") + async with _client(install.app()) as client: + created = ( + await client.post("/api/sessions", headers=install.headers("alice"), json={}) + ).json() + assert created["caused_by_actor_id"] == alice + # REST chat without a session opens the API channel's session. + await client.post( + "/api/chat", headers=install.headers("bob"), json={"message": "hi"}, + ) + await install.db.update_session_fields(created["id"], {"sdk_session_id": "native-1"}) + fork = (await client.post( + "/api/sessions/fork", headers=install.headers("bob"), + json={"source_session_id": created["id"]}, + )).json() + + api_session = await install.engine.sessions.get_last_session("api:default") + assert await install.session_cause_of(api_session) == bob + assert fork["caused_by_actor_id"] == bob + + +@pytest.mark.asyncio +async def test_a_tool_change_in_alices_session_is_for_alice_whoever_sends_the_turn( + install, monkeypatch, +): + """A shared session: every turn's agent work is for the session's starter.""" + install.use_fake_runtime(monkeypatch) + alice, bob = install.actor_id("alice"), install.actor_id("bob") + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _chat(client, install.headers("alice"), session_id, "first") + await _chat(client, install.headers("bob"), session_id, "second") + + assert [turn["cause"] for turn in install.turns_in(session_id)] == [alice, alice] + for turn in install.turns_in(session_id): + assert await install.events(turn["task_id"]) == [ + (None, "pending", session_id, install.system_actor_id, alice), + ] + # Each message still names who wrote it. + assert await install.user_rows(session_id) == [("first", alice), ("second", bob)] + assert install.engine.session_cause(session_id) == alice + + +@pytest.mark.asyncio +async def test_a_cron_sessions_tool_change_is_system_for_system(install, monkeypatch): + install.use_fake_runtime(monkeypatch) + await install.engine.run_cron(job_id="nightly", prompt="tidy the board") + + turn = install.turns[0] + system = install.system_actor_id + assert await install.session_cause_of(turn["session_id"]) == system + assert turn["cause"] == system + assert await install.events(turn["task_id"]) == [ + (None, "pending", turn["session_id"], system, system), + ] + + +@pytest.mark.asyncio +async def test_concurrent_sessions_of_two_people_keep_their_causes(install, monkeypatch): + install.use_fake_runtime(monkeypatch) + alice, bob = install.actor_id("alice"), install.actor_id("bob") + both_in_flight = asyncio.Barrier(2) + + async def _wait_for_the_other_turn(): + await asyncio.wait_for(both_in_flight.wait(), timeout=5) + + async with _client(install.app()) as client: + hers = await _new_session(client, install.headers("alice")) + his = await _new_session(client, install.headers("bob")) + install.runtime(hers).before_tool_call = _wait_for_the_other_turn + install.runtime(his).before_tool_call = _wait_for_the_other_turn + await asyncio.gather( + _chat(client, install.headers("alice"), hers, "from Alice"), + _chat(client, install.headers("bob"), his, "from Bob"), + ) + + for session_id, person in ((hers, alice), (his, bob)): + turn = install.turn_in(session_id) + assert turn["cause"] == person + assert await install.events(turn["task_id"]) == [ + (None, "pending", session_id, install.system_actor_id, person), + ] + + +@pytest.mark.asyncio +async def test_channel_input_with_no_person_has_no_cause(install, monkeypatch): + install.use_fake_runtime(monkeypatch) + await install.engine.run( + "chat-app", "hello from a chat app", source="telegram", actor=None, + ) + + turn = install.turn_in("chat-app") + assert await install.session_cause_of("chat-app") is None + assert turn["cause"] is None + assert await install.events(turn["task_id"]) == [ + (None, "pending", "chat-app", install.system_actor_id, None), + ] + + +@pytest.mark.asyncio +async def test_a_run_cause_applies_only_when_the_run_creates_the_session( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + alice, bob = install.actor_id("alice"), install.actor_id("bob") + system = install.db.system_actor + await install.engine.run("leg", "start", source="workflow", actor=system, cause=alice) + await install.engine.run("leg", "go on", source="workflow", actor=system, cause=bob) + await install.engine.run("unknown", "start", source="workflow", actor=system, cause=None) + + assert await install.session_cause_of("leg") == alice + assert [turn["cause"] for turn in install.turns_in("leg")] == [alice, alice] + assert await install.session_cause_of("unknown") is None + assert install.turn_in("unknown")["cause"] is None + + +@pytest.mark.asyncio +async def test_resume_after_restart_keeps_the_session_cause( + install, monkeypatch, tmp_path, +): + install.use_fake_runtime(monkeypatch) + alice = install.actor_id("alice") + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _chat(client, install.headers("bob"), session_id, "long job") + await install.db.update_session_fields(session_id, {"sdk_session_id": "native-1"}) + # A restart: a new connection knows the stored cause before any turn. + await install.db.close() + await install.db.connect() + assert install.engine.session_cause(session_id) == alice + + queue = tmp_path / "resume-after-restart" + queue.write_text(f"{session_id}\n") + with patch("nerve.agent.engine.RESUME_QUEUE_FILE", queue): + assert await install.engine.resume_enrolled_sessions() == 1 + + resumed = install.turns_in(session_id)[-1] + assert resumed["cause"] == alice + assert await install.events(resumed["task_id"]) == [ + (None, "pending", session_id, install.system_actor_id, alice), + ] + + +@pytest.mark.asyncio +async def test_a_fresh_engine_knows_every_stored_session_cause(install): + """After a restart, before any turn, nothing reads unknown for a known cause.""" + alice = install.actor_id("alice") + await install.db.create_session( + "hers", actor=Actor(actor_id=alice, kind="human"), + ) + await install.db.create_session( + "nobodys", actor=install.db.system_actor, cause=None, + ) + await install.db.close() + await install.db.connect() + engine = AgentEngine(install.config, install.db) + + assert engine.session_cause("hers") == alice + assert engine._build_tool_context("hers").cause_id == alice + assert engine._build_tool_context("hers").attribution == Attribution( + install.system_actor_id, alice, + ) + # A stored NULL stays NULL. + assert engine.session_cause("nobodys") is None + assert engine._build_tool_context("nobodys").cause_id is None + + +@pytest.mark.asyncio +async def test_the_session_cause_is_known_from_creation_and_from_a_read(install): + alice, bob = install.actor_id("alice"), install.actor_id("bob") + engine = install.engine + await install.db.create_session("new", actor=Actor(actor_id=alice, kind="human")) + assert engine.session_cause("new") == alice + + # A row that another writer adds after connect is known from its first read. + await install.db.db.execute( + "INSERT INTO sessions (id, title, source, status, caused_by_actor_id," + " created_at, updated_at) VALUES ('elsewhere', 'x', 'web', 'idle', ?," + " '2026-01-01', '2026-01-01')", + (bob,), + ) + await install.db.db.commit() + await install.db.get_session("elsewhere") + assert engine.session_cause("elsewhere") == bob + + await install.db.delete_session("new") + assert engine.session_cause("new") is None + + +@pytest.mark.asyncio +async def test_run_later_and_fork_keep_who_wrote_each_message(install): + alice = install.actor_id("alice") + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await client.post( + "/api/sessions/run-later", headers=install.headers("alice"), + json={"session_id": session_id, "message": "later", "delay": "none"}, + ) + await install.db.update_session_fields(session_id, {"sdk_session_id": "native-1"}) + fork = await client.post( + "/api/sessions/fork", headers=install.headers("bob"), + json={"source_session_id": session_id}, + ) + + assert await install.user_rows(session_id) == [("later", alice)] + assert await install.user_rows(fork.json()["id"]) == [("later", alice)] + + +# --------------------------------------------------------------------------- # +# Sessions that plans start # +# --------------------------------------------------------------------------- # + + +def _record_runs(install: _Install, monkeypatch) -> list[dict]: + runs: list[dict] = [] + ran = asyncio.Event() + + async def _run(*args, **kwargs): + runs.append(kwargs) + ran.set() + return "" + + monkeypatch.setattr(install.engine, "run", _run) + install.ran = ran # type: ignore[attr-defined] + return runs + + +async def _pending_plan(install: _Install, suffix: str) -> str: + task_id, plan_id = f"task-{suffix}", f"plan-{suffix}" + task_file = install.config.workspace / f"{task_id}.md" + task_file.write_text("# Plan task\n", encoding="utf-8") + await install.db.upsert_task( + task_id=task_id, file_path=task_file.name, title="Plan task", + status="pending", content="# Plan task\n", + ) + await install.db.create_plan( + plan_id=plan_id, task_id=task_id, content="Ship it", + session_id=f"planner-{suffix}", version=1, + ) + return plan_id + + +@pytest.mark.asyncio +async def test_rest_plan_approval_starts_a_session_for_the_approver(install, monkeypatch): + runs = _record_runs(install, monkeypatch) + plan_id = await _pending_plan(install, "rest") + async with _client(install.app()) as client: + response = await client.post( + f"/api/plans/{plan_id}/approve", headers=install.headers("bob"), + ) + assert response.status_code == 200 + await asyncio.wait_for(install.ran.wait(), timeout=5) + impl = response.json()["impl_session_id"] + assert await install.session_cause_of(impl) == install.actor_id("bob") + assert runs[0]["actor"] == install.db.system_actor + + +@pytest.mark.asyncio +async def test_tool_plan_approval_starts_a_session_for_the_approving_sessions_cause( + install, monkeypatch, +): + runs = _record_runs(install, monkeypatch) + plan_id = await _pending_plan(install, "tool") + alice = install.actor_id("alice") + # The agent approves the plan in a session that Alice started. + ctx = ToolContext( + session_id="alices-session", workspace=install.config.workspace, + db=install.db, config=install.config, engine=install.engine, + actor=install.db.system_actor, cause=alice, + ) + await plan_approve_handler(ctx, {"plan_id": plan_id}) + await asyncio.wait_for(install.ran.wait(), timeout=5) + impl = (await install.db.get_plan(plan_id))["impl_session_id"] + assert await install.session_cause_of(impl) == alice + assert runs[0]["actor"] == install.db.system_actor + + +@pytest.mark.asyncio +async def test_a_revision_request_creates_the_planner_session_for_the_requester( + install, monkeypatch, +): + _record_runs(install, monkeypatch) + plan_id = await _pending_plan(install, "revise") + async with _client(install.app()) as client: + response = await client.post( + f"/api/plans/{plan_id}/revise", headers=install.headers("bob"), + json={"feedback": "add tests"}, + ) + assert response.status_code == 200 + await asyncio.wait_for(install.ran.wait(), timeout=5) + assert await install.session_cause_of("planner-revise") == install.actor_id("bob") + + +# --------------------------------------------------------------------------- # +# The MCP endpoint # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_mcp_contexts_carry_the_system_actor_and_the_session_cause( + install, monkeypatch, +): + resolve = build_ctx_resolver(install.engine, SatelliteSessionResolver(install.db)) + rctx = SimpleNamespace(request=None, session=None) + + # A satellite session: its agent runs outside this instance. + satellite = await resolve(rctx) + assert await install.session_cause_of(satellite.session_id) is None + assert satellite.attribution == Attribution(install.system_actor_id, None) + + # A Codex session that this instance runs, started by Alice. + await install.engine.sessions.get_or_create( + "codex-session", source="web", + actor=Actor(actor_id=install.actor_id("alice"), kind="human"), + ) + monkeypatch.setattr( + "nerve.mcp_server.http._bound_identity_from_request", + lambda config, rctx: ("codex-session", {}), + ) + bound = await resolve(rctx) + assert bound.attribution == Attribution( + install.system_actor_id, install.actor_id("alice"), + ) + + +# --------------------------------------------------------------------------- # +# The actor directory # +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_the_actor_directory_names_every_id_in_every_registered_column( + install, monkeypatch, +): + install.use_fake_runtime(monkeypatch) + async with _client(install.app()) as client: + session_id = await _new_session(client, install.headers("alice")) + await _chat(client, install.headers("bob"), session_id, "make a task") + await client.post( + "/api/tasks", headers=install.headers("alice"), json={"title": "Direct"}, + ) + directory = { + row["id"] + for row in ( + await client.get("/api/actors", headers=install.headers("alice")) + ).json()["actors"] + } + + referenced: dict[tuple[str, str], set[str]] = {} + for table, column in ACTOR_COLUMNS: + async with install.db.db.execute( + f"SELECT DISTINCT {column} FROM {table} WHERE {column} IS NOT NULL" + ) as cursor: + referenced[(table, column)] = {row[0] async for row in cursor} + assert referenced[(table, column)] <= directory, (table, column) + # The scenario fills each column that this change adds. + for pair in ( + ("sessions", "caused_by_actor_id"), + ("task_events", "actor_id"), + ("task_events", "caused_by_actor_id"), + ): + assert referenced[pair], pair + + +# --------------------------------------------------------------------------- # +# External mode # +# --------------------------------------------------------------------------- # + + +def _b64url(value: object) -> str: + return base64.urlsafe_b64encode(json.dumps(value).encode()).rstrip(b"=").decode() + + +def _as(principal_id: str, name: str) -> dict: + payload = {"principal_id": principal_id, "profile": {"display_name": name}} + token = _b64url({"alg": "ES256", "typ": "JWT"}) + "." + _b64url(payload) + ".x" + return {ACTOR_CONTEXT_HEADER: token} + + +@pytest_asyncio.fixture +async def external_install(tmp_path, monkeypatch, wire_identity_store): + monkeypatch.setenv(AUTH_MODE_ENV, AUTH_MODE_EXTERNAL) + pin_auth_mode(AUTH_MODE_EXTERNAL) + config = NerveConfig.from_dict({ + "workspace": str(tmp_path / "ws"), + "codex": {"home_dir": str(tmp_path / "codex-home")}, + "auth": {"jwt_secret": _SECRET}, + }) + (config.workspace / "memory" / "tasks" / "active").mkdir(parents=True) + set_config(config) + database = Database(tmp_path / "nerve.db") + await database.connect() + try: + await bootstrap_identity(database, config) + wire_identity_store(database) + engine = AgentEngine(config, database) + init_deps(engine, database) + yield _Install(database, engine, config) + finally: + await database.close() + set_config(NerveConfig()) + + +@pytest.mark.asyncio +async def test_external_mode_records_principals_as_actor_and_cause( + external_install, monkeypatch, +): + install = external_install + install.use_fake_runtime(monkeypatch) + alice, bob = _as(ALICE_PRINCIPAL, "Alice"), _as(BOB_PRINCIPAL, "Bob") + async with _client(install.app()) as client: + task_id = (await client.post( + "/api/tasks", headers=alice, json={"title": "Hosted task"}, + )).json()["task"]["id"] + await client.post( + f"/api/tasks/{task_id}/move", headers=bob, json={"status": "deferred"}, + ) + session_id = await _new_session(client, bob) + await _chat(client, bob, session_id, "make a task") + await _chat(client, alice, session_id, "and one more") + + assert await install.events(task_id) == [ + (None, "pending", "web", ALICE_PRINCIPAL, ALICE_PRINCIPAL), + ("pending", "deferred", "web", BOB_PRINCIPAL, BOB_PRINCIPAL), + ] + assert await install.session_cause_of(session_id) == BOB_PRINCIPAL + for turn in install.turns_in(session_id): + assert turn["cause"] == BOB_PRINCIPAL + assert await install.events(turn["task_id"]) == [ + (None, "pending", session_id, install.system_actor_id, BOB_PRINCIPAL), + ] diff --git a/tests/test_db.py b/tests/test_db.py index dbf402ae9..c5045a8e5 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -939,6 +939,7 @@ async def _patch(self, db, tmp_path, monkeypatch, content, initial_tags="old"): monkeypatch.setattr(tasks_route, "get_deps", lambda: type("D", (), {"db": db})()) await tasks_route.update_task( "t1", tasks_route.TaskUpdateRequest(content=content), + actor=db.system_actor, ) return await db.get_task("t1") diff --git a/tests/test_plan_revise.py b/tests/test_plan_revise.py index 54bdf7630..06a7f2e57 100644 --- a/tests/test_plan_revise.py +++ b/tests/test_plan_revise.py @@ -52,12 +52,14 @@ async def get_or_create( metadata: dict | None = None, *, actor: Actor | None, + cause=None, ) -> dict: self.calls.append({ "session_id": session_id, "title": title, "source": source, "actor": actor, + "cause": cause, }) return {"id": session_id, "title": title or session_id, "source": source} @@ -82,12 +84,14 @@ async def run( source: str = "web", *, actor: Actor | None, + cause=None, ) -> None: self.runs.append({ "session_id": session_id, "user_message": user_message, "source": source, "actor": actor, + "cause": cause, }) self.run_event.set() @@ -124,6 +128,7 @@ async def test_dispatches_plan_update_prompt(self, db: Database, tmp_path): result = await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="be more specific about edge cases", + cause=None, ) # Wait for the background dispatch task to actually call engine.run. @@ -147,6 +152,7 @@ async def test_stores_feedback_and_writes_task_note(self, db: Database, tmp_path await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="needs more tests", + cause=None, ) await asyncio.wait_for(engine.run_event.wait(), timeout=1.0) @@ -169,6 +175,7 @@ async def test_routes_to_proposer_session(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="rev", + cause=None, ) await asyncio.wait_for(engine.run_event.wait(), timeout=1.0) @@ -191,6 +198,7 @@ async def test_falls_back_to_cron_planner_when_session_unknown( await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="rev", + cause=None, ) await asyncio.wait_for(engine.run_event.wait(), timeout=1.0) @@ -206,6 +214,7 @@ async def test_refuses_non_pending_plan(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="rev", + cause=None, ) # Nothing should have been dispatched. assert engine.runs == [] @@ -219,6 +228,7 @@ async def test_refuses_superseded_plan(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="rev", + cause=None, ) async def test_raises_plan_not_found(self, db: Database, tmp_path): @@ -228,6 +238,7 @@ async def test_raises_plan_not_found(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-does-not-exist", feedback="rev", + cause=None, ) assert engine.runs == [] @@ -242,6 +253,7 @@ async def test_raises_task_not_found(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback="rev", + cause=None, ) assert engine.runs == [] @@ -252,6 +264,7 @@ async def test_empty_feedback_raises_value_error(self, db: Database, tmp_path): await request_plan_revision( db=db, engine=engine, plan_id="plan-orig", feedback=" ", + cause=None, ) diff --git a/tests/test_task_board_api.py b/tests/test_task_board_api.py index d4f0b06bf..319150fd8 100644 --- a/tests/test_task_board_api.py +++ b/tests/test_task_board_api.py @@ -22,6 +22,8 @@ import pytest_asyncio from nerve.db import Database +from tests.actor_rows import ensure_actor_row +from tests.conftest import TEST_ACTOR @pytest.mark.asyncio @@ -55,6 +57,9 @@ async def setup(self, db: Database, tmp_path, bypass_auth): _skill_manager=None, ) init_deps(engine=engine, db=db) # type: ignore[arg-type] + # The routes store the request's actor on task events, and the + # foreign key needs its row. + await ensure_actor_row(db, TEST_ACTOR) app = FastAPI() app.include_router(tasks_router) diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 046168e28..5cdc33a17 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -545,7 +545,12 @@ export interface TaskEvent { /** null on the row recording the task's creation. */ from_status: string | null; to_status: string; + /** Origin of the change: `web`, a session id, `system` or `backfill`. */ actor: string; + /** Who made the change. `null` on rows written before it was recorded. */ + actor_id: string | null; + /** On whose behalf the change was made. `null` when unknown. */ + caused_by_actor_id: string | null; created_at: string; } diff --git a/web/src/components/Chat/ActorLabel.tsx b/web/src/components/Chat/ActorLabel.tsx index 0ec97f572..61fcb7a2e 100644 --- a/web/src/components/Chat/ActorLabel.tsx +++ b/web/src/components/Chat/ActorLabel.tsx @@ -41,6 +41,40 @@ function tooltip(verb: string, actorId: string, attribution: Attribution): strin return `${verb} ${name}`; } +/** + * Who made a change, and for whom: the actor's name, followed by + * "for " when the change was made on behalf of another person. That + * is the system actor (Nerve) doing work that a person started. A cause that + * is the actor itself, or the system actor, adds nothing. Without an actor id + * nothing renders, so a caller can show its own fallback for older rows. + */ +export function ActorAttribution({ actorId, causeId, verb = 'By' }: { + actorId: string | null | undefined; + causeId: string | null | undefined; + /** 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); + 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); + return ( + + {actor.system && + ); +} + export function ActorLabel({ actorId }: { actorId: string | null | undefined }) { const attribution = useAttribution(actorId); if (!actorId) return null; diff --git a/web/src/components/Tasks/TaskDetailBody.test.tsx b/web/src/components/Tasks/TaskDetailBody.test.tsx index 636fe5a92..c0f1e8aa5 100644 --- a/web/src/components/Tasks/TaskDetailBody.test.tsx +++ b/web/src/components/Tasks/TaskDetailBody.test.tsx @@ -8,6 +8,10 @@ import { TaskDetailBody } from './TaskDetailBody'; vi.mock('../../api/client', () => ({ api: { updateTask: vi.fn() }, + // The timeline's actor labels read the viewer through the auth store, + // which wires these on import. + setToken: vi.fn(), clearToken: vi.fn(), getToken: vi.fn(), + setUnauthorizedHandler: vi.fn(), })); /** diff --git a/web/src/components/Tasks/TaskTimeline.test.tsx b/web/src/components/Tasks/TaskTimeline.test.tsx new file mode 100644 index 000000000..1b8d51c7f --- /dev/null +++ b/web/src/components/Tasks/TaskTimeline.test.tsx @@ -0,0 +1,117 @@ +import { render, screen, waitFor } from '@testing-library/react'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ActorRef, TaskEvent } from '../../api/client'; + +vi.mock('../../api/client', () => ({ + api: { + listTaskEvents: vi.fn(), + listActors: vi.fn(), + }, + // ActorLabel reads the viewer through the auth store, which wires these. + setToken: vi.fn(), clearToken: vi.fn(), getToken: vi.fn(), + setUnauthorizedHandler: vi.fn(), +})); +vi.mock('../../stores/helpers/draftStorage', () => ({ clearAllDrafts: vi.fn() })); +vi.mock('../../stores/helpers/readStorage', () => ({ clearAllReads: vi.fn() })); + +const { api } = await import('../../api/client'); +const { useActorStore } = await import('../../stores/actorStore'); +const { TaskTimeline } = await import('./TaskTimeline'); +const { ActorAttribution } = await import('../Chat/ActorLabel'); + +const listTaskEvents = vi.mocked(api.listTaskEvents); +const listActors = vi.mocked(api.listActors); + +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 }; +} + +let nextId = 1; +function event(overrides: Partial): TaskEvent { + return { + id: nextId++, + task_id: 't1', + from_status: 'pending', + to_status: 'in_progress', + actor: 'web', + actor_id: null, + caused_by_actor_id: null, + created_at: '2026-01-01T00:00:00+00:00', + ...overrides, + }; +} + +function attributions(): string[] { + return [...document.querySelectorAll('[data-attribution="actor"]')] + .map((el) => el.textContent ?? ''); +} + +beforeEach(() => { + vi.clearAllMocks(); + useActorStore.getState().reset(); + listActors.mockResolvedValue({ + actors: [ + actorRef(ALICE, { display_name: 'Alice' }), + actorRef(BOB, { display_name: 'Bob' }), + actorRef(SYSTEM, { kind: 'system' }), + ], + }); +}); + +describe('TaskTimeline attribution', () => { + it('names the person, Nerve for a person, and Nerve on its own', async () => { + listTaskEvents.mockResolvedValue({ + events: [ + event({ from_status: null, to_status: 'pending', actor_id: ALICE, caused_by_actor_id: ALICE }), + event({ actor: 'sess-1', actor_id: SYSTEM, caused_by_actor_id: BOB }), + event({ to_status: 'done', actor: 'cron-1', actor_id: SYSTEM, caused_by_actor_id: SYSTEM }), + ], + }); + render(); + + await waitFor(() => expect(attributions()).toEqual(['Alice', 'Nerve for Bob', 'Nerve'])); + expect(screen.getByTitle('Changed by Nerve on behalf of Bob')).toBeInTheDocument(); + expect(screen.getByTitle('Changed by Alice')).toBeInTheDocument(); + // The origin text is not shown once the actor is known. + expect(screen.queryByText(/sess-1/)).toBeNull(); + }); + + it('keeps the origin text for rows without an actor id', async () => { + listTaskEvents.mockResolvedValue({ + events: [ + event({ from_status: null, to_status: 'pending', actor: 'backfill' }), + event({ actor: 'web' }), + event({ to_status: 'deferred', actor: 'system' }), + ], + }); + render(); + + expect(await screen.findByText(/· backfill/)).toBeInTheDocument(); + expect(screen.getByText(/· web/)).toBeInTheDocument(); + expect(screen.queryByText(/· system/)).toBeNull(); + expect(attributions()).toEqual([]); + }); +}); + +describe('ActorAttribution', () => { + it('shows Nerve alone when the cause is unknown', async () => { + render(); + expect(await screen.findByText('Nerve')).toBeInTheDocument(); + expect(screen.getByTitle('By Nerve itself — scheduled or autonomous work')).toBeInTheDocument(); + }); + + it('shows a person acting for themselves once', async () => { + render(); + expect(await screen.findByText('Bob')).toBeInTheDocument(); + expect(attributions()).toEqual(['Bob']); + }); + + it('renders nothing without an actor, so the caller can fall back', () => { + const { container } = render(); + expect(container).toBeEmptyDOMElement(); + }); +}); diff --git a/web/src/components/Tasks/TaskTimeline.tsx b/web/src/components/Tasks/TaskTimeline.tsx index 78f39376b..85bd3cd27 100644 --- a/web/src/components/Tasks/TaskTimeline.tsx +++ b/web/src/components/Tasks/TaskTimeline.tsx @@ -1,5 +1,6 @@ import { useEffect, useState } from 'react'; import { api, type TaskEvent } from '../../api/client'; +import { ActorAttribution } from '../Chat/ActorLabel'; import { StatusBadge } from './StatusControls'; import { formatTimeAgo } from '../../utils/dateGroups'; @@ -78,8 +79,19 @@ export function TaskTimeline({ taskId, currentStatus }: {
{formatTimeAgo(event.created_at)} - {event.actor && event.actor !== 'system' && ( - · {event.actor} + {event.actor_id ? ( + + {' · '} + + + ) : ( + // Rows written before the actor columns existed name only + // the origin of the change. + event.actor && event.actor !== 'system' && · {event.actor} )}