diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 9dee2a5025d..2559cd6d5c5 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -9,7 +9,7 @@ { "name": "doperpowers", "description": "Emerges by humans, Converges by Agent", - "version": "7.172.0", + "version": "7.174.0", "source": "./", "author": { "name": "SSFSKIM", diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 8a46f926b98..a602e5839cb 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "doperpowers", "description": "Emerges by humans, Converges by Agents", - "version": "7.172.0", + "version": "7.174.0", "author": { "name": "SSFSKIM", "email": "supremekim17@gmail.com" diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 06a1ef76f5d..96540332a76 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "doperpowers", - "version": "7.172.0", + "version": "7.174.0", "description": "One software-development methodology for coding agents, sized by the work: brainstorm, design, spec, TDD, review, ship — one spec that carries its plan of work as milestones: one milestone runs in order from the spec, several run a fresh subagent per milestone reviewed at each dependency frontier, and the same method runs unattended on the board loop.", "author": { "name": "SSFSKIM", diff --git a/docs/doperpowers/plans/2026-10-15-arkho-desk-c2-bridge.md b/docs/doperpowers/plans/2026-10-15-arkho-desk-c2-bridge.md index b9b1307f724..156735ea4eb 100644 --- a/docs/doperpowers/plans/2026-10-15-arkho-desk-c2-bridge.md +++ b/docs/doperpowers/plans/2026-10-15-arkho-desk-c2-bridge.md @@ -428,9 +428,12 @@ M1 left them (with fw1/fw2). Changed rows are marked **[M2]**. - **File:** `$SMINOS_HOME/bridge/.json`, mode 0600, written whole (temporary name, then rename) whenever it changes: - `{"override": {"model": , "effort": } | null, "delivered": {"": }}`. + `{"override": {"model": , "effort": } | null, "delivered": {"": }}`. + **[#229]** A mark is `/`, or the bare `` for a key naming no + project (`deliver`, below); the marks kept before keys named their project are bare, + and are kept as they are. - **Written from the mod's events:** `model` (sets `override`; null when both are null), - `delivered` with outcome `submitted` (raises the writer's seq), `resync` (replaces + `delivered` with outcome `submitted` (raises the key's mark), `resync` (replaces `override` and `delivered` with the mod's copy). - **[fw1] A failed write is not taken as kept.** The child holds the mod's copy in memory and writes it whole whenever it differs from what the file holds; what the file holds moves @@ -463,7 +466,7 @@ M1 left them (with fw1/fw2). Changed rows are marked **[M2]**. | `interrupt` | M2 **[M2]** **[fw1]** **[record]** | `{from?}` | `{turnId, stopped: [], running: []}`: `$.turn.abort` on the main turn, then `TaskStop` (`$.tool.call`) on each Bash call **attributable to that turn** and running at the abort whose result names the background task the abort moved it to (`running in background with ID: `; their results awaited at most 3 s). **[fw1]** Attributable: a main-loop call made during that turn, or a call of a foreground subagent spawned in it (nested foreground subagents inherit their parent's turn); a background subagent's calls never are. **[fw1]** `running` names each task whose TaskStop failed (an `isError` result or a rejection): the interrupt is incomplete when it is not empty. Idle: `{turnId: null, stopped: [], running: []}`. **[record]** A plugin's abort leaves the model no marker (Esc leaves `[Request interrupted by user]`), and a Bash call it moved to the background reads as still running there: so once the stops settle the interrupt appends two user rows. **[marker]** The first is Esc's marker whole, `[Request interrupted by user]`, one text block: a transcript's reader knows a turn's interrupt by a row whose whole text is the marker and by nothing else (the engine's own readers, 2.1.300 `He`/`pQn`, and MAWS's `INTERRUPT_MARKERS`), so the earlier one-row shape, `[Request interrupted by user from the desk]` with the task lines after it, read as no interrupt. The second is the note: `The interrupt came from the desk.` (`from: "desk"`, which the daemon's RPC sends) or `The interrupt came from outside the terminal.`, then a line per task: stopped, `Task , the Bash command the interrupt moved to the background, was then stopped: it is not running, and no notification will come for it.`; in `running`, `… could not be stopped: it may still be running.` The engine keeps a plugin's user row `isMeta: true` with `origin: {kind: "plugin", name}` whatever the plugin asks (2.1.300 `m6r`; it refuses any other user row): the model reads both rows, and no reader takes the note for a prompt. A marker a hook refuses is followed by no note. Both must be in before the turn ends: a next turn already queued takes its messages before it waits for any hook (2.1.300 `ir`/`cn`), so an append in `turn.complete` would miss its first request. The engine waits for a Bash call's result after a turn abort, so each Bash call of the turn being interrupted is held in the `tool.call` hook, after its result, until both rows are in (at most 10 s, `INTERRUPT_HOLD_MS`): they are kept before the moved call's result row, and the next request carries them after that result, the marker then the note (live, 2.1.302). An interrupt with no Bash call running appends at once after the abort, racing only the turn's own unwinding. A second interrupt of a turn being interrupted leaves the first's record alone (the engine refuses its abort). Each row is named by a `row-seen` notice (door `note`). The transcript mirrors the rows on one line as `interrupt: [Request interrupted by user] ` (a newline there draws as U+FFFD). Nothing for an idle interrupt or a refused abort | | `model` | M2 **[M2]** | `{model, effort?}` — `model` an id, a family alias, `"default"` or null; `effort` one of `low medium high xhigh max`, or null | `{model, effort}` now in force (null: the session's own). Rewritten on every **main-loop** `turn.step` request, subagents untouched; never `$.config.set`. A family alias (`haiku`, `sonnet`, `opus`, `fable`) is resolved through the session's `ANTHROPIC_DEFAULT__MODEL`; one it does not map is refused, naming the id to use. Emits `model`; kept by the child | | `compact` | M2 **[M2]** | `{instructions?}` | During the main turn `{queued: true, turnId}`: held, and run inside that turn's `turn.complete` (a second while one is held keeps the first). Idle `{started: true}`: run at once. Either way the outcome follows on the stream as `compact` events carrying this command's `id` | -| `deliver` | M2 **[M2]** **[fw1]** | `{key: {writer, seq}, text}` | `{submitted: true}` or `{duplicate: true}`. `text` is submitted with `$.prompt.submit({text, asUser: true})` (queued until the session is idle; not awaited). **The delivery key** is the log event's `(writer, seq)` — for a push carrying several events of one writer, the last. A key whose seq is at or below the highest already submitted for its writer is a duplicate and dropped (delivery goes per writer in seq order). The per-writer seqs are kept by the child across reloads; a /clear starts them anew (a new id is a new recipient). **[fw1]** The `delivered` event is emitted before the submission is made, so it is the next thing posted. Deduplication across a module reload is **best effort when the reload lands mid-submission**: a key submitted whose `delivered` had not reached the child is not kept, and a redelivery of it may submit again (a residual of at-least-once delivery, §3) | +| `deliver` | M2 **[M2]** **[fw1]** **[#229]** | `{key: {project, writer, seq}, text}` | `{submitted: true}` or `{duplicate: true}`. `text` is submitted with `$.prompt.submit({text, asUser: true})` (queued until the session is idle; not awaited). **The delivery key** is the log event's `(writer, seq)` — for a push carrying several events of one writer, the last. A key whose seq is at or below the highest already submitted for its writer is a duplicate and dropped (delivery goes per writer in seq order). The per-writer seqs are kept by the child across reloads; a /clear starts them anew (a new id is a new recipient). **[fw1]** The `delivered` event is emitted before the submission is made, so it is the next thing posted. Deduplication across a module reload is **best effort when the reload lands mid-submission**: a key submitted whose `delivered` had not reached the child is not kept, and a redelivery of it may submit again (a residual of at-least-once delivery, §3). **[#229]** The key names the log's project (its id): every project numbers a writer's events on its own, and a writer id is per session, not per project, so one writer delivering to one session from two projects reaches seqs the other project already passed — keyed by `(writer, seq)` alone, those were dropped as duplicates and settled, lost with no trace. Deduplication is per writer **within a project**, against the mark `/`; a key naming no project (a CLI older than this) is held to the bare `` mark, which is all such a CLI raised, so no kept mark is migrated, and a session whose bare mark is past a project's own count takes that project's scoped keys from the first. The CLI takes a `duplicate` as delivered (with scoped keys it was submitted before) but says it, to the sender and in `daemon.log`, naming the event and the session. A refusal (`refused`, below) keeps the key whole, project included, and the daemon reports only those naming its own project, or none | ##### Events (the mod's posts; the stream's lines) @@ -474,7 +477,7 @@ M1 left them (with fw1/fw2). Changed rows are marked **[M2]**. | `model` | M2 **[M2]** | `model, effort` — the choice now in force (null: the session's own) | | `compact` / `start` | M2 **[M2]** | `id` — the `compact` command's id | | `compact` / `end` | M2 **[M2]** | `id`, and one of: `tokensBefore, tokensAfter` (null when the engine recorded none); `skip` (a hook vetoed it); `error` (the engine refused it, as during a turn) | -| `delivered` | M2 **[M2]** (was reserved) | `key: {writer, seq}, outcome: "submitted" \| "duplicate" \| "refused"`, `reason?` — `refused` follows a `submitted` when the engine later dropped or rejected the prompt | +| `delivered` | M2 **[M2]** (was reserved), **[#229]** | `key: {project?, writer, seq}` (the command's key, whole), `outcome: "submitted" \| "duplicate" \| "refused"`, `reason?` — `refused` follows a `submitted` when the engine later dropped or rejected the prompt | | `kept` | M2 **[fw1]** **[fw2]** (the child's own) | `ok: false, error` when writing `.json` starts to fail (said once, however many batches it keeps failing; retried at the end of every batch); `ok: true` when a write works again after a failure | ##### Subscription (`GET /events`) diff --git a/docs/doperpowers/specs/2026-10-13-smns-project-log-design.md b/docs/doperpowers/specs/2026-10-13-smns-project-log-design.md index bc0b3e10fbc..74ca0d29306 100644 --- a/docs/doperpowers/specs/2026-10-13-smns-project-log-design.md +++ b/docs/doperpowers/specs/2026-10-13-smns-project-log-design.md @@ -200,6 +200,41 @@ terminal session the human opened, so reconciliation re-joins it only as a different process (another pid, or a later start). A resume re-joins, and the retire is not undone a moment later. +**Hand-off.** Sending an interactive session to the background +(`/background`, the agents view's ←) does not move its process: the +harness starts a job with `--resume --fork-session` — a new +session id, a new process, a new bridge endpoint — lets the interactive +process exit, and appends `{"type": "continued-in", "continuedInSessionId": +}` to the old transcript (its own `/resume` hides the old session by +it). The job is the same member: the old session is retired with `reason` +`continued` and `to` the new one, in its own file, and the projection +reads the new session as continuing it (`former`): what it has not set of +its own — alias, spawner, in-charge line, status, worktree — it carries on, +its chat cursor is copied, a seat that held the old session moves to it, +and a tag, `alias#short`, or full id naming the old session resolves to +it. The link is recorded when the job joins while the old process still +exits (the usual case: observed, it outlives the job's start by half a minute), else +when the old one's membership would end, and for a seat at every +reconciliation while the session it holds no longer runs (the harness +writes the record only once the job has started, so the job may join +first). The cursor is copied and the seat moved before the retirement is +recorded, so a hand-off cut short (a seat lock another command holds, a +hook out of time) leaves the old session joined and is done again whole. +The record is read as the harness reads it, the latest one with no turn +after it, so an old session resumed and worked in again is its own: it +re-joins, and a `joined` of its own after the `continued` retirement ends +the link, the job keeping what its own events hold. A `continued` +retirement holds against the exiting process as a `retire` does. + +**Names the harness makes up.** The alias of a seat-less session is its +harness name, unless that name is one the harness made up (`nameSource` +`derived`: the directory's name and one random byte, drawn again by every +process; `auto`: a title it wrote; `collision`: a given name it suffixed) +and the log already holds an alias for the member. A resume therefore +keeps `maws-4d` rather than becoming `maws-3f`, and a job keeps the alias +of the session it continues rather than its generated title. A name +someone gives (`/rename`, `--name`) renames the member as before. + There is no host seat. A seat that spawns another is recorded as its spawner (§6) and nothing more: it does not integrate the other's work, is not woken by its reports, and retiring it leaves the other running. The @@ -1789,6 +1824,23 @@ Dispatch scripts: one `--brief` each. and the hub verifies any kind. Date/Author: 2026-10-17, the desk's session-title change. +- Decision: a session sent to the background is the member it was (§2, + Hand-off), linked by the harness's own `continued-in` record and carried + in the log as a `retired` with `reason` `continued` and `to`; and a name + the harness made up never replaces an alias the log holds. The owner saw + a session's alias and socket change on every `/background` (MAWS, + 2026-10-17: `maws-fe` became "GUI AI agent workspace charter", a new + session, and the old one was retired `ended`), and a resume rename a + member (`maws-4d` → `maws-3f`, same session). Rejected: a member id + apart from the session id, which would re-key every record sminos and + the desk keep per session for a case the projection can fold; and a + stable socket path per member, which would let one endpoint answer for + two processes while both run. A retirement, not a new kind, + so an older reader (the desk's projection) reads the old session as + retired and the job as a member of its own, which is what it reads now. + The desk folding the two is MAWS's follow-up. + Date/Author: 2026-10-17, the hand-off change. + ## Outcomes & Retrospective 2026-10-14, at the end of execution and before the branch review. diff --git a/hooks/mods/bridge-state.tsx b/hooks/mods/bridge-state.tsx index 42c5a001b70..016c77eddc3 100644 --- a/hooks/mods/bridge-state.tsx +++ b/hooks/mods/bridge-state.tsx @@ -96,10 +96,12 @@ export type TranscriptRecord = { type: 'assistant' | 'user'; uuid: string; messa export type Check = { toolUseId: string; tool: string; agentId: string | null; input: unknown } /** One agent of the session as `$.agent.list()` gives it (the fields the bridge reads). */ export type ListedAgent = { id: string; description: string; type: string; status: string; parentId?: string; name?: string } +/** A log event's key: its project (absent from a CLI older than the project in the key), writer and seq. */ +export type DeliveryKey = { project?: string; writer: string; seq: number } /** A log event submitted as the human's words, whose row is still to be kept. */ -export type Submission = { key: { writer: string; seq: number }; text: string } +export type Submission = { key: DeliveryKey; text: string } /** A delivery the engine dropped after it was submitted, as `delivered {outcome: "refused"}` reports it. */ -export type Refusal = { key: { writer: string; seq: number }; reason: string } +export type Refusal = { key: DeliveryKey; reason: string } export type BridgeState = { /** @@ -158,7 +160,12 @@ export type BridgeState = { toolCalls: number /** The model and effort chosen through the bridge, if any. */ override?: Override - /** Per writer, the highest seq of a log event submitted: delivery goes in seq order, so any key at or below it was. */ + /** + * Per mark, the highest seq of a log event submitted: delivery goes in seq + * order, so any key at or below it was. A mark is `/`, or + * the bare writer for a key naming no project; kept marks are taken as + * they come, whichever they are. + */ delivered: Map /** * Whether the module's first hello was read: what the child keeps diff --git a/hooks/mods/bridge-stream.tsx b/hooks/mods/bridge-stream.tsx index b43f9ec5183..f7477605441 100644 --- a/hooks/mods/bridge-stream.tsx +++ b/hooks/mods/bridge-stream.tsx @@ -121,7 +121,7 @@ function track(e: BridgeEvent) { /** The refusals a post the child answered carried, in its own events or a resync's: the child keeps them. */ function acknowledge(batch: BridgeEvent[]) { - const name = (k: Refusal['key']) => `${k.writer}:${k.seq}` + const name = (k: Refusal['key']) => `${k.project ?? ''}|${k.writer}:${k.seq}` const carried = new Set( batch.flatMap((e) => e.type === 'resync' ? (e.refused as Refusal[]).map((r) => name(r.key)) : e.type === 'delivered' && e.outcome === 'refused' ? [name(e.key as Refusal['key'])] : [], @@ -235,9 +235,9 @@ export function adoptKept(kept: unknown) { state.override = next.model === null && next.effort === null ? undefined : next } if (typeof delivered === 'object' && delivered !== null) { - for (const [writer, seq] of Object.entries(delivered)) { - if (Number.isInteger(seq) && (seq as number) > (state.delivered.get(writer) ?? 0)) { - state.delivered.set(writer, seq as number) + for (const [mark, seq] of Object.entries(delivered)) { + if (Number.isInteger(seq) && (seq as number) > (state.delivered.get(mark) ?? 0)) { + state.delivered.set(mark, seq as number) } } } diff --git a/hooks/mods/bridge.tsx b/hooks/mods/bridge.tsx index aeb37361e7a..f88efb150a5 100644 --- a/hooks/mods/bridge.tsx +++ b/hooks/mods/bridge.tsx @@ -2,7 +2,7 @@ import type { EngineInterface, HttpResponse, On, PermissionRequestDecision, Sess import { ASK_TOOL, PLUGIN, observeAsks } from './ask' import { EFFORTS, RESPAWN_FIRST_MS, capped, isEffort, loopKey, message, parsed, state } from './bridge-state' -import type { Answer, Command, Decision, Override, Request, Steer, Stops } from './bridge-state' +import type { Answer, Command, Decision, DeliveryKey, Override, Request, Steer, Stops } from './bridge-state' import { adoptKept, adoptSeed, emit, endStep, forgetLive, poster, pump, releaseStepEnd, requeue, resyncEvent, stopChild } from './bridge-stream' import { CHECKS_KEPT, @@ -561,20 +561,30 @@ async function runCompaction($: EngineInterface, request: { id: string; instruct /** * A log event for this session, submitted as the human's words: queued until * the session is idle. Delivery is at least once and goes in seq order per - * writer, so a key at or below the writer's last submitted is a repeat, and - * dropped. + * writer within a project, so a key at or below its mark (the last submitted + * for that writer in that project) is a repeat, and dropped. Every project + * numbers a writer's events on its own, so the mark is `/`; + * a key naming no project (a CLI older than the project in the key) is held + * to the writer's bare mark, which is all such a CLI ever raised. */ function deliver($: EngineInterface, key: unknown, text: unknown): Answer { - const { writer, seq } = (typeof key === 'object' && key !== null ? key : {}) as { writer?: unknown; seq?: unknown } - if (typeof writer !== 'string' || writer === '' || !Number.isInteger(seq) || typeof text !== 'string') { - return { ok: false, error: 'deliver needs key {writer, seq} and text' } + const { project, writer, seq } = (typeof key === 'object' && key !== null ? key : {}) as { project?: unknown; writer?: unknown; seq?: unknown } + if ( + typeof writer !== 'string' || + writer === '' || + !Number.isInteger(seq) || + typeof text !== 'string' || + (project !== undefined && (typeof project !== 'string' || project === '')) + ) { + return { ok: false, error: 'deliver needs key {project?, writer, seq} and text' } } - const delivery = { writer, seq: seq as number } - if (delivery.seq <= (state.delivered.get(writer) ?? 0)) { + const delivery: DeliveryKey = { ...(project !== undefined && { project: project as string }), writer, seq: seq as number } + const mark = project === undefined ? writer : `${project}/${writer}` + if (delivery.seq <= (state.delivered.get(mark) ?? 0)) { emit({ type: 'delivered', key: delivery, outcome: 'duplicate' }) return { ok: true, result: { duplicate: true } } } - state.delivered.set(writer, delivery.seq) + state.delivered.set(mark, delivery.seq) emit({ type: 'delivered', key: delivery, outcome: 'submitted' }) // Its row, once kept, is named `entered` (bridge-rows.tsx, enteredRow). const submission = { key: delivery, text } @@ -584,7 +594,7 @@ function deliver($: EngineInterface, key: unknown, text: unknown): Answer { } const refused = (reason: string) => { state.submissions = state.submissions.filter((s) => s !== submission) - $.ui.log(`bridge: the delivery ${writer}:${delivery.seq} did not enter: ${reason}`, { to: 'debug' }) + $.ui.log(`bridge: the delivery ${mark}:${delivery.seq} did not enter: ${reason}`, { to: 'debug' }) emit({ type: 'delivered', key: delivery, outcome: 'refused', reason }) } // Not awaited: it resolves when the prompt's turn starts, after any running turn. diff --git a/package.json b/package.json index d5f877cb0b8..3a138e8fd36 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "doperpowers", - "version": "7.172.0", + "version": "7.174.0", "description": "Doperpowers skills and runtime bootstrap for coding agents", "type": "module", "keywords": [ diff --git a/skills/sminos/SKILL.md b/skills/sminos/SKILL.md index 71796d7285a..2e8f6d82056 100644 --- a/skills/sminos/SKILL.md +++ b/skills/sminos/SKILL.md @@ -23,7 +23,12 @@ Members are siblings. There is no host seat: the host is the human, and a session that spawns another is recorded as its spawner (`by`) and nothing more — it does not supervise or integrate the other's work, and retiring it leaves the other running. An interactive session the human -closes leaves the family on its own and rejoins when it is resumed. +closes leaves the family on its own and rejoins when it is resumed. One +the human sends to the background (`/background`, ←) becomes a new +session; it goes on as the same member — alias, in-charge line, status, +chat position, seat — and a tag or id naming the old session reaches it. +A name the harness makes up (a directory label with a random suffix, a +title it writes) never renames a member; a name someone gives does. Everything that crosses sessions is an event in the project's append-only log: a message to a member, a session's `` to its @@ -172,7 +177,7 @@ in-charge line, then `· now:` the status and `· by` the spawner. `list` hides retired and gone seats behind a `+N hidden` tail; `--all` or `--state` shows them. `--json` carries every member with `session`, `short`, `machine`, `state`, `charge`, `charge_source`, `by`, `status`, -`worktree`, and `seat`. `chart` and `tui` draw the same words (`●` busy, +`worktree`, `seat`, and `former` (the sessions it continues). `chart` and `tui` draw the same words (`●` busy, `○` idle, `◐` waiting, `■` stopped, `✕` gone, `◌` vacant, `⊘` retired). ## Spawning siblings diff --git a/skills/sminos/scripts/sminos.py b/skills/sminos/scripts/sminos.py index 4ffc2bbf64f..583756c4569 100644 --- a/skills/sminos/scripts/sminos.py +++ b/skills/sminos/scripts/sminos.py @@ -508,6 +508,11 @@ class Project(NamedTuple): # `interactive` too (harness 2.1.299: `kind: "interactive", entrypoint: # "sdk-cli"`); its entrypoint is what tells it apart, and it is no member. HEADLESS_ENTRYPOINT = "sdk" +# The name sources the harness makes up itself (harness 2.1.300): `derived`, +# the directory's name and one random byte drawn again by every process; +# `auto`, a title it wrote; `collision`, a given name it suffixed because +# another live session held it. Any other source is a name someone gave. +GENERATED_NAMES = ("derived", "auto", "collision") _PROJECTS = {} _HOMES = set() _VIEWS = {} # project id → View, per invocation; a lifecycle append drops it @@ -1059,6 +1064,7 @@ class Member(NamedTuple): by: str worktree: str seat: dict | None # the seat record this session fills, when it is a seat on this machine + former: tuple = () # the sessions it continues (lineage), newest first def member_json(m, project=None): @@ -1066,7 +1072,7 @@ def member_json(m, project=None): d = {"alias": m.alias, "session": m.session, "short": m.short, "name_source": m.name_source, "machine": m.machine, "local": m.local, "state": m.state, "state_age": m.state_age, "charge": m.charge, "charge_source": m.charge_source, "by": m.by, "status": m.status, - "worktree": m.worktree, "seat": public_seat(m.seat) if m.seat else None} + "worktree": m.worktree, "seat": public_seat(m.seat) if m.seat else None, "former": list(m.former)} if project is not None: d = {"project": project.name, "project_id": project.id, **d} return d @@ -1212,8 +1218,13 @@ def _reconcile(project, events, local, by_session): `interactive` or `bg`: another tool's headless run is no member. And the absence of the process ends what its presence began: a seat-less member of this machine whose session no longer runs is retired, reason `ended`, - and re-joins if it runs again. A seat's lifecycle stays with retire, fill, - and resume.""" + and re-joins if it runs again — unless the harness handed it to a + background job, which it continues (hand_off). A seat's lifecycle stays + with retire, fill, and resume, but for the same hand-off: a seat whose + session no longer runs and names a job it continues in moves to it. The + job may have joined before the harness wrote that record, and a hand-off + a held seat lock cut short is left to be tried again, so every + reconciliation looks.""" latest, joined = {}, {} for e in events: if e["kind"] in MEMBERSHIP and e.get("session") and writer_session(e["writer"]) == e["session"]: @@ -1225,6 +1236,10 @@ def _reconcile(project, events, local, by_session): if (latest[sid] != "joined" or sid in by_session or sid in local or not is_self(str(e.get("machine") or ""), sid) or session_running(sid)): continue + successor = continued_in(sid) + if successor: + out += hand_off(project, sid, successor) + continue def ended(evs, sid=sid): return (latest_membership(evs) == "joined" and not session_running(sid) @@ -1236,6 +1251,10 @@ def ended(evs, sid=sid): if (latest.get(sid) == "joined") if peer else (sid in latest): continue out += join_on_sight(project, sid, seat, peer, bool(peer)) + for sid, ev in local.items(): + seat = ev.get("seat") + if seat and "peer" not in ev and seat["status"] != "retired" and (successor := continued_in(sid)): + out += hand_off(project, sid, successor) return out @@ -1281,11 +1300,13 @@ def retirement_holds(events, peer): re-joining it on sight would undo the retire at the next reading verb. The record shows a different process when its pid differs from the one the retirement recorded, or when it started after the retirement (a - resume opens a new process, so a resumed session re-joins, §2). A - retirement for any other reason (`ended`: its process was not seen) does - not hold against a running process.""" + resume opens a new process, so a resumed session re-joins, §2). So does + a hand-off (`continued`): the process that handed its session to a + background job runs on for a while as it exits. A retirement for any + other reason (`ended`: its process was not seen) does not hold against a + running process.""" last = next((e for e in reversed(events) if e["kind"] in MEMBERSHIP), None) - if not last or last["kind"] != "retired" or last.get("reason") != "retire": + if not last or last["kind"] != "retired" or last.get("reason") not in ("retire", "continued"): return False peer = peer or {} if last.get("pid") not in (None, "") and str(last["pid"]) != str(peer.get("pid")): @@ -1302,7 +1323,12 @@ def join_on_sight(project, sid, seat, peer, running): event at all; a seat-less session retired by `retire` only once a new process runs it (retirement_holds). Checked again under the session file's lock, the seat record re-read there, so a concurrent retire cannot - be undone. Shared by reconciliation and the hook (§2, §4).""" + be undone. Shared by reconciliation and the hook (§2, §4). + + A seat-less session joins under the alias it already holds (held_alias); + a background job the harness forked from an interactive session that is + still exiting joins as that member's continuation (hand_off), under its + alias.""" if not joinable(seat, peer): return [] @@ -1315,12 +1341,196 @@ def guard(evs): fresh = reload_seat(seat["seat_id"]) if seat else None return not (seat and (fresh is None or fresh["status"] == "retired")) + out = [] if seat: fields = joined_fields(seat["alias"], "seat", seat["cwd"], seat["worktree"], seat["role"], seat["seat_id"]) else: - fields = joined_fields(str(peer.get("name") or sid[:8]), str(peer.get("nameSource") or "derived"), - str(peer.get("cwd") or "")) - return lifecycle(project, sid, "joined", guard=guard, required=False, **fields) + held = own_alias(_own_events(project, sid), sid) + if str(peer.get("kind") or "") == "bg" and not held[0]: + old, old_peer = predecessor(project, sid) + if old: + out += hand_off(project, old, sid, old_peer) + held = own_alias(_own_events(project, old), old) + alias, source = held_alias(*held, peer, sid) + fields = joined_fields(alias or sid[:8], source or "derived", str(peer.get("cwd") or "")) + return out + lifecycle(project, sid, "joined", guard=guard, required=False, **fields) + + +def held_alias(logged, logged_source, peer, session): + """(alias, name_source) a seat-less session goes by: its harness name + when someone gave it, else the alias the log already holds for it + (`logged`, own_alias: its own, or the one of the session it continues), + else its harness name; ("", "") for none. A name the harness made up + (GENERATED_NAMES) changes with each process — a resume draws a new + suffix, a background job gets a title — so it never replaces an alias + already held: the address stays the one its peers know.""" + name = str((peer or {}).get("name") or "") + source = str((peer or {}).get("nameSource") or "derived") + if logged == session[:8]: # the placeholder a nameless session joined under + logged = "" + if name and (source not in GENERATED_NAMES or not logged): + return name, source + return (logged, logged_source) if logged else ("", "") + + +def own_alias(events, session): + """(alias, name_source) of the latest joined or renamed among a session's + own events; ("", "") for none, or the placeholder (its short id).""" + named = [e for e in events if e["kind"] in ("joined", "renamed") and e.get("alias")] + if not named or str(named[-1]["alias"]) == session[:8]: + return "", "" + return str(named[-1]["alias"]), str(named[-1].get("name_source") or "") + + +# How much of a transcript's end is read for its hand-off record: the harness +# appends it as the session leaves (a few KiB of bookkeeping follow), and +# reads 1 MiB back for it itself. +HANDOFF_REACH = 1 << 20 +_HANDOFFS = {} # session -> ((path, size, mtime_ns), successor): valid while the file is unchanged + + +def continued_in(session): + """The session `session` continues in, when the harness sent it to the + background (`/background`, the agents view's ←): that starts a job as a + fork, a new session id and process, lets the old process exit, and + appends `{"type": "continued-in", "continuedInSessionId": }` to the + old transcript — the record its own /resume reads to hide the old + session (harness 2.1.300). Read as the harness reads it: the latest such + record with no turn after it, so an old session resumed and worked in + again is its own again. "" when there is none.""" + path = transcript_path(session) + try: + st = os.stat(path) if path else None + except OSError: + st = None + if st is None: + return "" + key = (path, st.st_size, st.st_mtime_ns) + hit = _HANDOFFS.get(session) + if hit and hit[0] == key: + return hit[1] + found = "" + try: + with open(path, "rb") as f: + for _, line in _lines_backward(f, _complete_end(f), HANDOFF_REACH): + if b'"continued-in"' not in line and b'"user"' not in line and b'"assistant"' not in line: + continue + r = _record(line) or {} + if r.get("type") == "continued-in": + to = str(r.get("continuedInSessionId") or "") + found = to if UUID_RE.match(to) and to != session else "" + break + if is_turn_record(r): + break + except OSError: + return "" + _HANDOFFS[session] = (key, found) + return found + + +def is_turn_record(r): + """Whether a transcript record is part of a turn, as the harness tells + one after a hand-off: a finished assistant message (not an API error), or + a prompt the human typed.""" + if r.get("type") == "assistant": + msg = r.get("message") if isinstance(r.get("message"), dict) else {} + return r.get("isApiErrorMessage") is not True and isinstance(msg.get("stop_reason"), str) + if r.get("type") != "user" or r.get("isMeta"): + return False + content = (r.get("message") or {}).get("content") if isinstance(r.get("message"), dict) else None + return (isinstance(content, str) and bool(content.strip())) or ( + isinstance(content, list) and any(isinstance(b, dict) and b.get("type") == "text" for b in content)) + + +def predecessor(project, session): + """(session, harness record) of the session `session` continues: a + running interactive session of this machine in the project (only one can + be sent to the background), still exiting, whose transcript names + `session` (continued_in); else a session a seat of the project holds that + no longer runs, with no record; (None, None) for none. A seat-less one + that has exited already is reconciliation's to find, as its membership + ends.""" + for rec in peer_records(): + old = rec.get("sessionId") + if (isinstance(old, str) and old != session and UUID_RE.match(old) and rec.get("kind") == "interactive" + and peer_running(rec) and (p := resolve_project(str(rec.get("cwd") or ""))) is not None + and p.id == project.id and continued_in(old) == session): + return old, rec + for s in project_seats(project): + old = s["current"] + if old and old != session and s["status"] != "retired" and not session_running(old) \ + and continued_in(old) == session: + return old, None + return None, None + + +def hand_off(project, old, new, old_peer=None): + """Record that the member `old` stood for continues in session `new` + (continued_in): its chat cursor carried to `new`; a seat that held `old` + moved to `new`; `old` retired, reason `continued`, naming `new` (and the + process it saw, which may still be exiting: retirement_holds); and `new`, + if it already joined under a name the harness made up, renamed to the + alias `old` held. The projection (lineage) carries the rest: by, the + in-charge line, status, worktree. Returns the events appended. + + Ordered so a hand-off cut short is finished by the next one: the cursor + merge is idempotent, the retirement comes only after the seat moved (a + seat whose lock another command holds leaves `old` joined, and the next + reconciliation tries the whole of it again), and the rename runs while + `old` stands retired to `new`, whoever retired it.""" + def still_joined(evs): + return latest_membership(evs) == "joined" + + try: + carried = cursor_load(project, old) + mine = writer_id(old) + seqs = [e["seq"] for e in _own_events(project, old) if e["writer"] == mine] + if seqs: + carried[mine] = max(carried.get(mine, 0), max(seqs)) + if carried: + cursor_save(project, new, carried) + except OSError as e: + warn("carrying %s's chat cursor to %s failed: %s" % (old[:8], new[:8], e)) + out = [] + seat = next((s for s in seats() if s["current"] == old and (p := seat_project(s)) and p.id == project.id), None) + if seat: + try: + out += follow_seat(seat, old, new) + except ResumeRefused: + return out + fields = {"pid": old_peer.get("pid")} if old_peer else {} + out += lifecycle(project, old, "retired", guard=still_joined, required=False, by="", reason="continued", + to=new, **fields) + theirs = _own_events(project, old) + last = next((e for e in reversed(theirs) if e["kind"] in MEMBERSHIP), None) + if not last or last["kind"] != "retired" or last.get("reason") != "continued" or last.get("to") != new: + return out + alias, source = own_alias(theirs, old) + own = _own_events(project, new) + if alias and latest_membership(own) == "joined": + mine_alias, mine_source = own_alias(own, new) + if (not mine_alias or mine_source in GENERATED_NAMES) and (mine_alias, mine_source) != (alias, source): + out += lifecycle(project, new, "renamed", guard=still_joined, required=False, + alias=alias, name_source=source) + return out + + +def follow_seat(seat, old, new): + """Move a seat from `old` to the session it continues in, under the + seat's lock: the same occupant goes on, so nothing else of the record + changes. A seat that no longer holds `old` is left as it is; a seat + whose lock another command holds raises ResumeRefused, for the caller + to try again later. Returns the events appended.""" + locks = lock_seat(seat, refusal=ResumeRefused) + try: + fresh = reload_seat(seat["seat_id"]) + if fresh is None or fresh["current"] != old: + return [] + row = agent_row(session_id=new) + meta_set(seat["seat_id"], {"current": new, "short": str((row or {}).get("id") or ""), "updated": now()}) + return join_seat(reload_seat(seat["seat_id"])) + finally: + unlock(locks) def latest_across_writers(events): @@ -1339,15 +1549,23 @@ def _projection(events): latest joined (by seq, in its own file), the first non-empty `by`, its latest rename, published state, and published title (its own file), its in-charge events (for charge_of), and its status: each writer's latest by - seq, the later ts between writers (§3).""" + seq, the later ts between writers (§3). And its hand-offs: `to`, the + session it continues in (the latest `retired` reason `continued` of its + own, unless it joined again since: a session resumed after it was handed + off is its own member, and the job keeps what its own events hold), and + `continues`, the session that names it so (lineage).""" info = {} + + def row(sid): + return info.setdefault(sid, {"latest": None, "joined": None, "by": "", "renamed": None, "state": None, + "titled": None, "in_charge": [], "status": [], "to": "", "continues": ""}) + for e in events: sid, kind = e.get("session"), e["kind"] if kind not in LIFECYCLE or not isinstance(sid, str) or not sid: continue own = writer_session(e["writer"]) == sid - r = info.setdefault(sid, {"latest": None, "joined": None, "by": "", "renamed": None, "state": None, - "titled": None, "in_charge": [], "status": []}) + r = row(sid) if kind in MEMBERSHIP or kind in ("renamed", "state", "titled"): if not own: continue @@ -1355,19 +1573,53 @@ def _projection(events): r["latest"] = e if kind == "joined": r["joined"] = e + r["to"] = "" if not r["by"] and e.get("by"): r["by"] = str(e["by"]) + elif e.get("reason") == "continued" and UUID_RE.match(str(e.get("to") or "")): + r["to"] = str(e["to"]) else: r[kind] = e elif kind == "in-charge": r["in_charge"].append(e) else: r["status"].append(e) + for sid, r in list(info.items()): + if r["to"] and r["to"] != sid: + row(r["to"])["continues"] = sid for r in info.values(): r["status"] = latest_across_writers(r["status"]) return info +def lineage(sid, info): + """The sessions `sid` continues (hand_off), newest first: the one it was + forked from, that one's, and so on.""" + out, cur = [], (info.get(sid) or {}).get("continues") + while cur and cur != sid and cur not in out: + out.append(cur) + cur = (info.get(cur) or {}).get("continues") + return out + + +def logged_alias(sid, info): + """(alias, name_source) the log holds for a session (own_alias, read off + the projection): its own latest name when someone gave it, else the + oldest alias along the sessions it continues, a given name there first. + ("", "") for none.""" + held = ("", "") + for s in [sid] + lineage(sid, info): + r = info.get(s) + named = (r["renamed"] or r["joined"] or {}) if r else {} + alias, source = str(named.get("alias") or ""), str(named.get("name_source") or "") + if not alias or alias == s[:8]: + continue + if source not in GENERATED_NAMES: + return alias, source + held = alias, source + return held + + def charge_of(member, events, title=True, published=""): """(line, source): what a member is in charge of, the first present of (§5) the latest `set` in-charge line, the name the human gave it, the @@ -1376,8 +1628,10 @@ def charge_of(member, events, title=True, published=""): none is. A set or derived line written empty is no line. `events` is the log, or any part of it holding the member's in-charge events, each writer's in seq order. With title=False the transcript is not read (the - daemon's tick has no use for the line).""" - mine = [e for e in events if e["kind"] == "in-charge" and e.get("session") == member.session] + daemon's tick has no use for the line). A line set for a session the + member continues (`former`) is the member's.""" + sessions = (member.session,) + tuple(member.former) + mine = [e for e in events if e["kind"] == "in-charge" and e.get("session") in sessions] for source in ("set", "name", "derived", "title"): if source == "name": line = member.alias if member.name_source == "user" else "" @@ -1407,17 +1661,20 @@ def ts_age(ts): return max(0.0, time.time() - then) -def _member(sid, r, seat, peer, local, title=True): +def _member(sid, r, seat, peer, local, title=True, info=None): """One row of the projection: what the log says of the session, and, on - this machine, its seat record and its process.""" + this machine, its seat record and its process. A session that continues + others (lineage, in `info`) is the member they stood for: what it has + not set of its own — its alias (held_alias), its spawner, its status, + its worktree, its in-charge line — it carries on from them.""" + info = info if info is not None else ({sid: r} if r else {}) + former = tuple(lineage(sid, info)) + chain = [info[s] for s in (sid,) + former if info.get(s)] j = (r or {}).get("joined") or {} - alias, name_source = str(j.get("alias") or ""), str(j.get("name_source") or "") - if r and r["renamed"]: - alias, name_source = str(r["renamed"].get("alias") or alias), str(r["renamed"].get("name_source") or name_source) if seat: alias, name_source = seat["alias"], "seat" - elif peer and peer.get("name"): - alias, name_source = str(peer["name"]), str(peer.get("nameSource") or name_source or "derived") + else: + alias, name_source = held_alias(*logged_alias(sid, info), peer, sid) machine = str(j.get("machine") or "") local = local or bool(seat) or not machine or is_self(machine, sid) age = None @@ -1431,15 +1688,20 @@ def _member(sid, r, seat, peer, local, title=True): state_word = peer_state(peer) else: state_word = "stopped" if transcript_path(sid) else "gone" - status = (r or {}).get("status") + status = max((x["status"] for x in chain if x["status"]), key=lambda e: e["ts"], default=None) + worktree = next((str((x["joined"] or {}).get("worktree") or "") for x in chain + if (x["joined"] or {}).get("worktree")), "") m = Member( session=sid, short=sid[:8], alias=alias or sid[:8], name_source=name_source, machine=host_name() if local else machine, local=local, state=state_word, state_age=age, charge="", charge_source="", status=str(status.get("text") or "") if status else (seat["now"] if seat else ""), - by=(r or {}).get("by") or "", worktree=seat["worktree"] if seat else str(j.get("worktree") or ""), - seat=seat) - charge, source = charge_of(m, (r or {}).get("in_charge") or [], title, published_title(r)) + by=next((x["by"] for x in chain if x["by"]), ""), worktree=seat["worktree"] if seat else worktree, + seat=seat, former=former) + # charge_of reads each writer's events in seq order, and one writer (a + # remote operator) may have set a line for each session of the chain. + charge, source = charge_of(m, sorted((e for x in chain for e in x["in_charge"]), + key=lambda e: (e["writer"], e["seq"])), title, published_title(r)) return m._replace(charge=charge, charge_source=source) @@ -1480,6 +1742,10 @@ def view_of(project, all_seats=None, titles=True): local = _local_sessions(project, all_seats) events = log_read(project) appended = _reconcile(project, events, local, by_session) + if appended: # a hand-off may have moved a seat to the session it continues in + all_seats = seats() + by_session = seats_by_session(project, all_seats) + local = _local_sessions(project, all_seats) events = events + appended info = _projection(events) internal = internal_pids() @@ -1487,21 +1753,23 @@ def view_of(project, all_seats=None, titles=True): joined = [(sid, r) for sid, r in info.items() if r["latest"] and r["latest"]["kind"] == "joined" and sid not in forks] joined.sort(key=lambda x: (x[1]["joined"]["ts"], x[0]), reverse=True) - out = [_member(sid, r, by_session.get(sid), local.get(sid, {}).get("peer"), sid in local, titles) + out = [_member(sid, r, by_session.get(sid), local.get(sid, {}).get("peer"), sid in local, titles, info) for sid, r in joined] sessions = {m.session for m in out} rest = sorted((s for s in all_seats if (not s["current"] or s["current"] not in sessions) and (p := seat_project(s)) and p.id == project.id), key=lambda s: s["updated"], reverse=True) - rows = [_member(s["current"], info.get(s["current"]), s, None, True, titles) if s["current"] else + rows = [_member(s["current"], info.get(s["current"]), s, None, True, titles, info) if s["current"] else Member(session="", short="", alias=s["alias"], name_source="seat", machine=host_name(), local=True, state=state(s), state_age=None, charge="", charge_source="", status=s["now"], by="", worktree=s["worktree"], seat=s) for s in rest] # Seat-less sessions the log holds as retired, by alias: what a tag that - # names one is told. + # names one is told. One that continues in another session is not: its + # name is the continuation's now. retired = [(str((r["renamed"] or r["joined"]).get("alias") or sid[:8]), sid[:8], str(r["latest"].get("reason") or "")) for sid, r in info.items() - if r["latest"] and r["latest"]["kind"] == "retired" and r["joined"] and sid not in by_session] + if r["latest"] and r["latest"]["kind"] == "retired" and r["joined"] and sid not in by_session + and r["latest"].get("reason") != "continued"] view = View(out, rows, retired, events, info, len(appended)) if titles: _VIEWS[project.id] = view @@ -1639,7 +1907,7 @@ def push_event(project, member, session, e, events, cursor): if member.seat is None and latest_membership(_own_events(project, session)) == "retired": return "recorded" return push_member(member, frame_of(project, e, events, session, cursor), not e.get("wake"), session=session, - event=e) + event=e, project=project.id) def compose_chat_frame(project_name, ref, sender, target, unread, text): @@ -1687,16 +1955,18 @@ def is_all(tag): def match_members(name, short, projects): """The (project, member) pairs a tag's name answers to among the projects' members: their alias (with `short`, an alias whose short starts so), their - short (`@short`), or their session id. A retired member answers to nothing.""" + short (`@short`), or their session id — a session they continue (`former`) + answering as theirs does. A retired member answers to nothing.""" found = [] for p in projects: for m in members(p): if m.state == "retired": continue + ids = (m.session,) + m.former if short: - hit = m.alias == name and m.short.startswith(short) + hit = m.alias == name and any(s[:8].startswith(short) for s in ids) else: - hit = name in (m.alias, m.short, m.session) + hit = name == m.alias or any(name in (s, s[:8]) for s in ids) if hit: found.append((p, m)) return found @@ -1807,11 +2077,12 @@ def __init__(self, message, code=EXIT_UNKNOWN): self.code = code -def push_member(member, text, tagged, session=None, event=None): +def push_member(member, text, tagged, session=None, event=None, project=None): """Push one frame to a member: to the live session (push_live: its bridge, - keyed by `event`, the log event the frame carries; its supervisor; its - inbox socket); for a seat, a resume when it is stopped and the message is - addressed to it (`tagged`). A session sminos did not spawn is never + keyed by `event`, the log event the frame carries, and `project`, the id + of the project whose log holds it; its supervisor; its inbox socket); for + a seat, a resume when it is stopped and the message is addressed to it + (`tagged`). A session sminos did not spawn is never resumed: it is the human's to reopen, and reads the chat when it is back. Given `session`, a seat that now holds another session is not this message's recipient: delivery goes by session id, so a replacement cannot @@ -1822,7 +2093,7 @@ def push_member(member, text, tagged, session=None, event=None): if not peer: return "recorded" try: - control_module().push_live(peer, session, text, event) + control_module().push_live(peer, session, text, event, project) return "sent" except SendFailed as e: return "sent?" if e.phase == "after" else "failed:%s" % e @@ -1847,7 +2118,7 @@ def unreachable(s): peer = peer_for_session(fresh["current"]) if peer: try: - control_module().push_live(peer, fresh["current"], text, event) + control_module().push_live(peer, fresh["current"], text, event, project) return "sent" except SendFailed as e: return "sent?" if e.phase == "after" else "failed:%s" % e @@ -2494,19 +2765,15 @@ def _on_sight(project, session): def _member_alias(project, session, seat, peer): """Joins a joinable session the log does not hold as a member yet (a live process is evidence of activation), and returns the alias its messages to - the host carry: its seat's, else its harness name, else the last name the - log gave it. None when it is no member.""" + the host carry: its seat's, else the one it goes by (held_alias over the + last name the log gave it). None when it is no member.""" own = _own_events(project, session) if latest_membership(own) != "joined": join_on_sight(project, session, seat, peer, True) own = _own_events(project, session) if latest_membership(own) != "joined": return None - alias = (seat or {}).get("alias") or str((peer or {}).get("name") or "") - if not alias: - named = [e for e in own if e["kind"] in ("joined", "renamed") and e.get("alias")] - alias = str(named[-1]["alias"]) if named else session[:8] - return alias + return (seat or {}).get("alias") or held_alias(*own_alias(own, session), peer, session)[0] or session[:8] def export_live(cwd, session, records): @@ -3059,7 +3326,9 @@ def find_target(q): and "" when the name named the seat: a verb that re-reads the seat under its lifecycle lock refuses when the seat no longer holds that session (locked_fresh), so a refill in between cannot take the bound session's - place. + place. The seat and its session are read after the reconcile: a name of + the session the seat held binds the job it went on in, when the + reconcile handed the seat off to one. A full id — a seat id or session id — names its seat directly, and so does `group/alias` when the qualifier is a group and no project. Any other @@ -3076,8 +3345,13 @@ def exact(kind, res): if kind == "ambiguous": die(ambiguous_seat(q, res), EXIT_UNKNOWN) if kind == "ok": + # The reconcile can hand the seat off to the job its session went on + # in, so the record is read again after it, and a name that bound the + # session the seat held binds the session it holds now. + bound = res["current"] == q and res["seat_id"] != q reconcile(seat_project(res)) # the verb reads the project; no liveness probe needed - return ("seat", res, q if res["current"] == q and res["seat_id"] != q else "") + res = reload_seat(res["seat_id"]) or res + return ("seat", res, res["current"] if bound else "") return None def group_seat(group, alias): @@ -3098,10 +3372,11 @@ def group_seat(group, alias): for p in projects: ms, rows = project_view(p) for m in ms + rows: + ids = (m.session,) + m.former if m.session else () if short: - hit = bool(m.short) and m.alias == name and m.short.startswith(short) + hit = m.alias == name and any(s[:8].startswith(short) for s in ids) else: - hit = name == m.alias or (bool(m.session) and name == m.session) + hit = name == m.alias or name in ids # One seat is one target, though its session joined one project # and its record now resolves to another (a cwd that moved). key = m.seat["seat_id"] if m.seat else m.session @@ -3118,7 +3393,7 @@ def group_seat(group, alias): if not found: return (proj and not short and group_seat(proj, name)) or ("none",) project, m = found[0] - return ("seat", m.seat, m.session if short or name == m.session else "") if m.seat else ("member", project, m) + return ("seat", m.seat, m.session if short or name in (m.session,) + m.former else "") if m.seat else ("member", project, m) def ambiguous_seat(q, hits): @@ -3244,14 +3519,16 @@ def agents_refresh(): def forget_caches(): """Forget everything cached for the invocation — the agents rows, the peer - registry, process start times, resolved projects and their views — so a - long-lived process (the chart TUI, the daemon's loop) sees the fleet as it - is now rather than as it was at launch. Nothing is read until asked.""" + registry, process start times, transcript paths, resolved projects and + their views — so a long-lived process (the chart TUI, the daemon's loop) + sees the fleet as it is now rather than as it was at launch. Nothing is + read until asked.""" global _AGENTS_LOADED, _PEERS, _SELF _AGENTS_LOADED = False _PEERS = None _SELF = None _PSTART.clear() + _TRANSCRIPTS.clear() _PROJECTS.clear() _HOMES.clear() _VIEWS.clear() @@ -3838,6 +4115,12 @@ def status_for_state(state): return "working" +# session -> its transcript's path, per invocation: reconciliation and state() +# ask for the same stopped sessions' each pass. Only a hit is kept — spawn and +# resume poll for a transcript to appear. +_TRANSCRIPTS = {} + + def transcript_path(session_id): """Munging-agnostic: the harness mangles the cwd into the project-dir name, so glob for the transcript by its unique session id instead. A session's @@ -3849,8 +4132,14 @@ def transcript_path(session_id): one written last.""" if not session_id: return "" + hit = _TRANSCRIPTS.get(session_id) # one lookup: the session plane's threads ask while the uplink forgets + if hit: + return hit hits = glob.glob(os.path.join(home_dir(), ".claude", "projects", "*", session_id + ".jsonl")) - return max(hits, key=_mtime) if hits else "" + if not hits: + return "" + _TRANSCRIPTS[session_id] = path = max(hits, key=_mtime) + return path def transcript_rows(session_id): @@ -5036,7 +5325,9 @@ def default_from(explicit): The harness exports CLAUDE_CODE_SESSION_ID to Bash tools, so an AGENT is identified by it: the alias of the seat whose `current` is that session, - else the harness session name from the peer registry, else `session:`. + else the alias it goes by in the working directory's project (held_alias: + its harness name unless that is a made-up one the log already holds + another alias for), else `session:`. An agent may not claim to be the operator: `--from human` with a session id in the environment is refused. Without a session id (a real terminal) the default stays `human`.""" @@ -5051,10 +5342,10 @@ def default_from(explicit): for s in seats(): if s["current"] == sid: return s["alias"] - for r in peer_records(): - if r.get("sessionId") == sid and r.get("name"): - return str(r["name"]) - return "session:%s" % sid[:8] + peer = next((r for r in peer_records() if r.get("sessionId") == sid), None) + project = resolve_project(os.getcwd()) + held = own_alias(_own_events(project, sid), sid) if project is not None else ("", "") + return held_alias(*held, peer, sid)[0] or "session:%s" % sid[:8] def wait_socket_turn(s, marker, was_idle, guard): @@ -5171,7 +5462,8 @@ def resume(locks): was_idle = (normalize_state(row) not in ("working", "blocked")) if row else (peer.get("status") != "busy") try: if delivery: - control_module().push_live(peer, s["current"], delivery.payload(True), delivery.queue[-1]) + control_module().push_live(peer, s["current"], delivery.payload(True), delivery.queue[-1], + delivery.project.id) else: send_frame(sock, text) except SendFailed as e: @@ -5348,7 +5640,7 @@ def send_to_member(project, m, sender, via, text, msg_id): print("sent to %s (delivered by a concurrent delivery)" % m.alias) return try: - control_module().push_live(peer, m.session, delivery.payload(True), delivery.queue[-1]) + control_module().push_live(peer, m.session, delivery.payload(True), delivery.queue[-1], project.id) except SendFailed as e: if e.phase == "after": delivery.settle() diff --git a/skills/sminos/scripts/sminos_bridge.py b/skills/sminos/scripts/sminos_bridge.py index 870c8d04916..3141cd5c79d 100644 --- a/skills/sminos/scripts/sminos_bridge.py +++ b/skills/sminos/scripts/sminos_bridge.py @@ -132,12 +132,12 @@ def _no_kept(): def load_kept(session): """What a child kept for the session: {"override": {model, effort} | None, - "delivered": {writer: seq}}, and, once a command sent for a desk RPC was - answered, "rpcs": {nonce: the answer's op, ok, result, error}; once the - engine dropped a delivery the mod had submitted, "refused": [{"key": - {writer, seq}, "reason"}], oldest first, the last REFUSED_KEPT, which the - daemon reports (sminos_outcomes.report_refused); nothing kept reads as - none.""" + "delivered": {mark: seq}}, a mark as _mark gives it, and, once a command + sent for a desk RPC was answered, "rpcs": {nonce: the answer's op, ok, + result, error}; once the engine dropped a delivery the mod had submitted, + "refused": [{"key": {project?, writer, seq}, "reason"}], oldest first, the + last REFUSED_KEPT, which the daemon reports + (sminos_outcomes.report_refused); nothing kept reads as none.""" try: with open(kept_path(session)) as f: d = json.load(f) @@ -163,7 +163,15 @@ def _refusal(r): """Whether `r` is a refusal as the child keeps it.""" key = r.get("key") if isinstance(r, dict) else None return isinstance(key, dict) and isinstance(key.get("writer"), str) and type(key.get("seq")) is int \ - and isinstance(r.get("reason"), str) + and isinstance(key.get("project", ""), str) and isinstance(r.get("reason"), str) + + +def _mark(key): + """The mark a delivery key raises, as the mod's: every project numbers a + writer's events on its own, so `/`, or the bare writer + for a key naming no project (a CLI older than the project in the key).""" + project, writer = key.get("project"), key.get("writer") + return "%s/%s" % (project, writer) if isinstance(project, str) else writer # ------------------------------------------------------------------ the client @@ -274,12 +282,11 @@ def _refused(kept, key, reason): last REFUSED_KEPT, its reason cut to REASON_MAX; as it was for a key that is not one.""" key = key if isinstance(key, dict) else {} - writer, seq = key.get("writer"), key.get("seq") + key = {k: key[k] for k in ("project", "writer", "seq") if k in key} refused = kept.get("refused") or [] - if not isinstance(writer, str) or type(seq) is not int \ - or any(r["key"] == {"writer": writer, "seq": seq} for r in refused): + if not _refusal({"key": key, "reason": ""}) or any(r["key"] == key for r in refused): return kept - entry = {"key": {"writer": writer, "seq": seq}, "reason": str(reason or "")[:REASON_MAX]} + entry = {"key": key, "reason": str(reason or "")[:REASON_MAX]} return {**kept, "refused": (refused + [entry])[-REFUSED_KEPT:]} @@ -678,9 +685,9 @@ def _apply(self, e): self._keep({**self.kept, "override": o if o["model"] is not None or o["effort"] is not None else None}) elif kind == "delivered" and e.get("outcome") == "submitted": key = e.get("key") or {} - writer, seq = key.get("writer"), key.get("seq") - if isinstance(writer, str) and type(seq) is int and seq > self.kept["delivered"].get(writer, 0): - self._keep({**self.kept, "delivered": {**self.kept["delivered"], writer: seq}}) + mark, seq = _mark(key), key.get("seq") + if isinstance(mark, str) and type(seq) is int and seq > self.kept["delivered"].get(mark, 0): + self._keep({**self.kept, "delivered": {**self.kept["delivered"], mark: seq}}) elif kind == "delivered" and e.get("outcome") == "refused": # The engine dropped what the mod submitted: kept with its key, # for the daemon's next delivery pass to report `failed` (DC8.G5). @@ -1237,9 +1244,10 @@ def cmd_bridge(a): def watch_session(q): """The session a member name names here, or exit 4 saying why not. A full - session id is taken as given.""" + session id is taken as given, or, when the harness sent it to the + background and it no longer runs, as the session it continues in.""" if S.UUID_RE.match(q): - return q + return successor(q) target = S.find_target(q) if target[0] == "none": S.die("no member matching '%s'" % q, S.EXIT_UNKNOWN) @@ -1255,6 +1263,16 @@ def watch_session(q): return m.session +def successor(session): + """The session that goes on for `session`: itself while it runs or was + never handed off, else the end of its chain of hand-offs (continued_in).""" + seen = {session} + while not S.session_running(session) and (nxt := S.continued_in(session)) and nxt not in seen: + seen.add(nxt) + session = nxt + return session + + def _first_line(text, width=160): line = str(text or "").strip().split("\n", 1)[0] return line if len(line) <= width else line[:width - 1] + "…" @@ -1359,7 +1377,7 @@ def feed(self, e): self.line("· compaction %s" % ("started" if phase == "start" else _compact_outcome(e))) elif kind == "delivered" and e.get("outcome") != "duplicate": k = e.get("key") or {} - self.line("· delivered %s:%s %s" % (k.get("writer"), k.get("seq"), e.get("outcome"))) + self.line("· delivered %s:%s %s" % (_mark(k), k.get("seq"), e.get("outcome"))) elif kind == "gap": self.line("· %s events were lost on the way from the session" % e.get("dropped")) elif kind == "kept": @@ -1407,6 +1425,15 @@ def _reconnect_window(): return 5.0 +def _follow_window(): + """How long a watch waits for the bridge of the session its session was + handed to: that process starts, loads its plugins, and only then serves.""" + try: + return float(os.environ.get("SMINOS_BRIDGE_FOLLOW") or 60) + except ValueError: + return 60.0 + + def cmd_watch(a): session = watch_session(a.member) path = endpoint(session) @@ -1447,6 +1474,16 @@ def say(text): except (OSError, http.client.HTTPException, ValueError): pass if ended: + # Sent to the background: the conversation goes on in another + # session, whose bridge answers once its process is up. + nxt = S.continued_in(session) + deadline = time.time() + _follow_window() + while nxt and time.time() < deadline and probe(endpoint(nxt)) is None: + time.sleep(0.2) + if nxt and probe(endpoint(nxt)) is not None: + say("· sent to the background, continued in %s; following" % nxt[:8]) + session, path, ended, pid = nxt, endpoint(nxt), False, None + continue say("· the session ended") return if lost and probe(path) is not None: diff --git a/skills/sminos/scripts/sminos_control.py b/skills/sminos/scripts/sminos_control.py index dc36a09aff7..727ff490fa3 100644 --- a/skills/sminos/scripts/sminos_control.py +++ b/skills/sminos/scripts/sminos_control.py @@ -9,8 +9,8 @@ `push_live` is a delivery's last hop to a live session, for the log's delivery and `send` alike: the bridge, which submits the text as the human's -words, keyed by the log event (a key it has already submitted is dropped, so -delivery stays at least once); else, for a human's event to a --bg job, the +words, keyed by the log event and its project (a key it has already +submitted is dropped, so delivery stays at least once); else, for a human's event to a --bg job, the supervisor's `reply` (human-origin input); else the inbox frame. Each path carries the frame the inbox would, so the model reads the same addressing. """ @@ -72,12 +72,16 @@ def post_command(path, body, wait_ms=COMMAND_WAIT_MS): c.close() -def deliver(session, event, text): - """The bridge's submission of a log event: True once submitted (or found - already submitted), None when the session has no bridge, False when the - bridge refused it (nothing was submitted). Raises SendFailed("after") - when the bridge took it and gave no answer: it may have been submitted.""" - body = {"op": "deliver", "key": {"writer": event["writer"], "seq": event["seq"]}, "text": text} +def deliver(session, project, event, text): + """The bridge's submission of a log event of the project `project` (its + id): True once submitted (or found already submitted), None when the + session has no bridge, False when the bridge refused it (nothing was + submitted). Raises SendFailed("after") when the bridge took it and gave no + answer: it may have been submitted. The key names the project, since + every project numbers a writer's events on its own. A duplicate is taken + as delivered, since it was submitted before, but said: were it not, the + event would be lost with no other trace.""" + body = {"op": "deliver", "key": {"project": project, "writer": event["writer"], "seq": event["seq"]}, "text": text} try: r = post_command(B.endpoint(session), body) except Unanswered as e: @@ -86,6 +90,10 @@ def deliver(session, event, text): return None status, out = r if status == 200 and isinstance(out, dict) and out.get("ok"): + if (out.get("result") or {}).get("duplicate"): + line = "the bridge of session %s took %s as already delivered" % (session, S.event_ref(event)) + S.warn(line) + S.daemon_log(line) return True if status == 504: raise S.SendFailed("after", "the bridge's session did not answer") @@ -163,12 +171,12 @@ def supervisor_reply(short, text): # ------------------------------------------------------------------ delivery -def push_live(peer, session, text, event=None): +def push_live(peer, session, text, event=None, project=None): """A delivery's last hop to a live session, `event` the last log event the - text carries (None for a frame no log holds). Raises SendFailed as - send_frame does: "before" when nothing reached the session, "after" when - it may have.""" - if event is not None and deliver(session, event, text): + text carries (None for a frame no log holds) and `project` the id of the + project whose log holds it. Raises SendFailed as send_frame does: + "before" when nothing reached the session, "after" when it may have.""" + if event is not None and deliver(session, project, event, text): return if event is not None and event.get("kind") == "human" and bg_job(peer): took = supervisor_reply(bg_job(peer), text) diff --git a/skills/sminos/scripts/sminos_outcomes.py b/skills/sminos/scripts/sminos_outcomes.py index 1963825c848..c0ec04a6c81 100644 --- a/skills/sminos/scripts/sminos_outcomes.py +++ b/skills/sminos/scripts/sminos_outcomes.py @@ -285,7 +285,13 @@ def report_refused(project, view): for name in names: sid, ext = os.path.splitext(name) if ext == ".json" and sid in view.info: - refused = bridge.load_kept(sid).get("refused") + # A bridge keeps the refusals of every project that delivered to + # its session, and each project numbers a writer's events on its + # own: one naming another project is that project's to report. + # One naming none (a CLI older than the project in the key) is + # matched here, as it always was. + refused = [r for r in bridge.load_kept(sid).get("refused") or [] + if r["key"].get("project", project.id) == project.id] if refused: found[sid] = refused if not found: diff --git a/tests/mods/bridge.test.ts b/tests/mods/bridge.test.ts index df66455615e..4b4c0e052d0 100644 --- a/tests/mods/bridge.test.ts +++ b/tests/mods/bridge.test.ts @@ -1500,6 +1500,40 @@ describe('deliver', () => { ]) }) + test("each project numbers a writer's events on its own: the mark is per project, and a key naming none is held to the bare mark", async () => { + const h = harness() + await h.ready() + const p1 = { project: 'p1', writer: 'mac/s1', seq: 98 } + const p2 = { project: 'p2', writer: 'mac/s1', seq: 98 } + expect(await h.command({ op: 'deliver', key: p1, text: 'p1 98' })).toMatchObject({ result: { submitted: true } }) + expect(await h.command({ op: 'deliver', key: p2, text: 'p2 98' })).toMatchObject({ result: { submitted: true } }) + expect(await h.command({ op: 'deliver', key: { ...p2, seq: 6 }, text: 'p2 6' })).toMatchObject({ result: { duplicate: true } }) + expect(await h.command({ op: 'deliver', key: { writer: 'mac/s1', seq: 6 }, text: 'bare 6' })).toMatchObject({ result: { submitted: true } }) + expect(await h.command({ op: 'deliver', key: { writer: 'mac/s1', seq: 5 }, text: 'bare 5' })).toMatchObject({ result: { duplicate: true } }) + expect(await h.command({ op: 'deliver', key: { ...p1, project: '' }, text: 'x' })).toMatchObject({ ok: false }) + expect(h.calls.filter((c) => c.op === 'submit').map((c) => (c.args as { text: string }).text)).toEqual(['p1 98', 'p2 98', 'bare 6']) + expect(h.ofType('delivered').map((e) => [e.key, e.outcome])).toEqual([ + [p1, 'submitted'], + [p2, 'submitted'], + [{ ...p2, seq: 6 }, 'duplicate'], + [{ writer: 'mac/s1', seq: 6 }, 'submitted'], + [{ writer: 'mac/s1', seq: 5 }, 'duplicate'], + ]) + }) + + test("a hello's kept marks are honored as they come: a bare mark guards only keys naming no project, a scoped one only its project", async () => { + const h = harness() + await h.start() + // #229: a bare mark a writer raised in one project, kept before keys named theirs. + h.children[0]!.write(hello(SID, SOCK, { override: null, delivered: { 'mac/s1': 98, 'p1/mac/s2': 4 } })) + await h.settle() + expect(await h.command({ op: 'deliver', key: { project: 'p2', writer: 'mac/s1', seq: 6 }, text: 'p2 6' })).toMatchObject({ result: { submitted: true } }) + expect(await h.command({ op: 'deliver', key: { writer: 'mac/s1', seq: 98 }, text: 'bare 98' })).toMatchObject({ result: { duplicate: true } }) + expect(await h.command({ op: 'deliver', key: { project: 'p1', writer: 'mac/s2', seq: 4 }, text: 'p1 4' })).toMatchObject({ result: { duplicate: true } }) + expect(await h.command({ op: 'deliver', key: { project: 'p3', writer: 'mac/s2', seq: 4 }, text: 'p3 4' })).toMatchObject({ result: { submitted: true } }) + expect(h.calls.filter((c) => c.op === 'submit').map((c) => (c.args as { text: string }).text)).toEqual(['p2 6', 'p3 4']) + }) + test('a deliver with no key or no text is refused', async () => { const h = harness() await h.ready() @@ -1612,7 +1646,7 @@ describe("a delivered prompt's row (G6)", () => { await h.settle() h.posts[0]!.release() await h.settle() - const key = { writer: 'hub/m1', seq: 8 } + const key = { project: 'p1', writer: 'hub/m1', seq: 8 } h.children[0]!.write(line({ id: 'c1', op: 'deliver', key, text: 'dropped' })) await h.settle() ;(await untilRefusal(h)).fail() // the refusal never reaches the child diff --git a/tests/sminos/run-sminos-tests.sh b/tests/sminos/run-sminos-tests.sh index 1b77a3d04a5..6b56276217e 100755 --- a/tests/sminos/run-sminos-tests.sh +++ b/tests/sminos/run-sminos-tests.sh @@ -3086,7 +3086,7 @@ spec = importlib.util.spec_from_file_location("sminos", sys.argv[1]); mod = impo project, seat_id, mode, record = mod.find_project(sys.argv[2]), sys.argv[3], sys.argv[4], sys.argv[5] member = next(m for m in mod.members(project) if m.session == seat_id) down = os.path.join(os.path.dirname(record), "push-down") -def fake_push(member, text, tagged, session=None, event=None): +def fake_push(member, text, tagged, session=None, event=None, project=None): body = text.split("\n", 1)[1] time.sleep(0.05) if body == "ten" and os.path.exists(down): @@ -3624,6 +3624,274 @@ assert_not_contains "$OUT" "headless" "a session of another kind (a tool's headl assert_file_absent "$PFLOG/testhost.$HL.jsonl" "and is never joined" rm -f "$HOME/.claude/sessions/pf-headless.json" +# A session sent to the background (/background, the agents view's ←) is +# forked by the harness into a job under a new session id, the interactive +# process exiting, and its old transcript ends in a `continued-in` record +# naming the job. The member goes on in the job: under its alias, with its +# in-charge line, status, chat position, and seat, and what named the old +# session names the job. +handoff_tx() { # old new — the old transcript as the harness leaves it at a hand-off + printf '%s\n' '{"type":"user","message":{"role":"user","content":"keep going"}}' \ + '{"type":"assistant","uuid":"h0000001","message":{"role":"assistant","stop_reason":"end_turn","content":[{"type":"text","text":"done"}]}}' \ + "{\"type\":\"continued-in\",\"timestamp\":\"2026-10-17T21:01:52.408Z\",\"sessionId\":\"$1\",\"continuedInSessionId\":\"$2\"}" \ + '{"type":"last-prompt","lastPrompt":"keep going"}' > "$HOME/.claude/projects/pl-proj/$1.jsonl" +} +bg_peer() { # name session nameSource — a background job's harness record, with an inbox of its own + [ -S "$TEST_ROOT/in-$1.sock" ] || start_inbox "$1" + printf '{"pid":%s,"sessionId":"%s","name":"%s","nameSource":"%s","kind":"bg","status":"idle","cwd":"%s","messagingSocketPath":"%s"}\n' \ + "$SOCK_PID" "$2" "$1" "$3" "$R" "$TEST_ROOT/in-$1.sock" > "$HOME/.claude/sessions/pf-$1.json" +} +HO=4f00001d-0000-4000-8000-00000000004f +HN=5a00001e-0000-4000-8000-00000000005a +pf_peer handy "$HO" idle derived +run "$SMINOS" list +CLAUDE_CODE_SESSION_ID="$HO" run "$SMINOS" charge "owns the hand-off" +run "$SMINOS" status handy "mid-flight" +CLAUDE_CODE_SESSION_ID="$HO" run "$SMINOS" chat +handoff_tx "$HO" "$HN" +bg_peer handyjob "$HN" auto # the job registers while the interactive process still exits +: > "$HOME/.claude/projects/pl-proj/$HN.jsonl" +hook_run SessionStart "$HN" "$R" "$HOME/.claude/projects/pl-proj/$HN.jsonl" startup +assert_rc 0 "$HOOK_RC" "the job's SessionStart hook exits 0" +assert_contains "$(events_of "$HO")" "status retired:continued " "the job's start retires the session it continues, reason continued" +assert_contains "$(tail -1 "$PFLOG/testhost.$HO.jsonl")" "\"reason\": \"continued\", \"to\": \"$HN\"" "naming the job" +assert_contains "$(cat "$PFLOG/testhost.$HN.jsonl")" '"alias": "handy", "name_source": "derived"' "which joins under the member's alias" +run "$SMINOS" list +assert_not_contains "$OUT" "handyjob" "the job does not list under the title the harness gave it" +assert_equals "$(json_row handy session)" "$HN" "the member is the job now" +assert_equals "$(json_row handy former)" "[\"$HO\"]" "continuing the session it was forked from" +assert_equals "$(json_row handy charge)" "owns the hand-off" "with its in-charge line" +assert_equals "$(json_row handy status)" "mid-flight" "and its status" +CLAUDE_CODE_SESSION_ID="$HN" run "$SMINOS" chat +assert_contains "$OUT" "(no new messages)" "and its place in the chat" +run "$SMINOS" say "@handy over to you" +assert_rc 0 "$RC" "a tag of the alias is accepted" +assert_contains "$(inbox handyjob)" "over to you" "and reaches the job" +assert_not_contains "$(inbox handy)" "over to you" "not the process still exiting" +run "$SMINOS" say "@handy#${HO:0:6} by the old short" +assert_rc 0 "$RC" "alias#short of the session it continues still names the member" +assert_contains "$(inbox handyjob)" "by the old short" "and reaches the job" +run "$SMINOS" daemon tick +run "$SMINOS" list +assert_not_contains "$(events_of "$HO")" "continued joined" "the exiting process does not re-join while it runs" +rm -f "$HOME/.claude/sessions/pf-handy.json" # it exits +run "$SMINOS" list +assert_not_contains "$(events_of "$HO")" "ended" "nor does its exit end anything" +assert_equals "$(events_of "$HN")" "joined " "and the job holds the alias with no rename" +FOLLOWED="$(python3 - "$REPO_ROOT/skills/sminos/scripts" "$HO" "$HN" <<'PY_FOLLOW' +import json, sys +sys.path.insert(0, sys.argv[1]) +import sminos, sminos_bridge +old, new = sys.argv[2], sys.argv[3] +print(sminos_bridge.watch_session(old) == new, end=" ") +# An old session resumed and worked in again is its own: the record no longer ends it. +with open(sminos.transcript_path(old), "a") as f: + f.write(json.dumps({"type": "user", "message": {"role": "user", "content": "one more thing"}}) + "\n") +print(sminos.continued_in(old) == "") +PY_FOLLOW +)" +assert_equals "$FOLLOWED" "True True" "a full id of a handed-off session names the job (watch, the controls); a resumed one is its own again" +rm -f "$HOME/.claude/sessions/pf-handyjob.json" +# A job seen before its predecessor's record: a member of its own until the +# interactive process exits, when the hand-off is read as its membership ends. +LO=5b00001f-0000-4000-8000-00000000005b +LN=5c000020-0000-4000-8000-00000000005c +pf_peer tardy "$LO" idle derived +run "$SMINOS" list +bg_peer tardyjob "$LN" auto +run "$SMINOS" list +assert_equals "$(json_row tardyjob session)" "$LN" "a job seen before any hand-off names it is a member of its own" +handoff_tx "$LO" "$LN" +rm -f "$HOME/.claude/sessions/pf-tardy.json" +run "$SMINOS" list +assert_equals "$(events_of "$LO")" "joined retired:continued " "the exit is read as the hand-off, not an end" +assert_equals "$(events_of "$LN")" "joined renamed " "the job is renamed to the alias it carries on" +assert_equals "$(json_row tardy session)" "$LN" "and is the member" +assert_not_contains "$OUT" "tardyjob" "no longer listed under its title" +rm -f "$HOME/.claude/sessions/pf-tardyjob.json" +# A seat that held the session moves to the job. +SO=5e000022-0000-4000-8000-00000000005e +SN=5f000023-0000-4000-8000-00000000005f +run "$SMINOS" seat add pf-repo warden --session "$SO" +pf_peer warden "$SO" idle user +run "$SMINOS" list +handoff_tx "$SO" "$SN" +bg_peer wardenjob "$SN" auto +run "$SMINOS" list +assert_equals "$(field "$SO" current)" "$SN" "a seat that held the session moves to the job" +assert_equals "$(json_row warden session)" "$SN" "which stands for the seat" +assert_not_contains "$OUT" "wardenjob" "under the seat's alias alone" +assert_contains "$(events_of "$SO")" "retired:continued" "the session it held is retired, continued" +rm -f "$HOME/.claude/sessions/pf-warden.json" "$HOME/.claude/sessions/pf-wardenjob.json" +"$SMINOS" remove pf-repo/warden >/dev/null +# A seat whose job joined before the harness wrote the record: the hand-off +# is read once the record is there and the old process has exited. +KO=6a000024-0000-4000-8000-00000000006a +KN=6b000025-0000-4000-8000-00000000006b +run "$SMINOS" seat add pf-repo keeper --session "$KO" +pf_peer keeper "$KO" idle user +run "$SMINOS" list +bg_peer keeperjob "$KN" auto +run "$SMINOS" list +assert_equals "$(events_of "$KN")" "joined " "a seat's job seen before the record joins as a member of its own" +handoff_tx "$KO" "$KN" +rm -f "$HOME/.claude/sessions/pf-keeper.json" +run "$SMINOS" list +assert_equals "$(field "$KO" current)" "$KN" "once the record is there and the seat's process gone, the seat moves to the job" +assert_contains "$(events_of "$KO")" "retired:continued" "and the session it held is retired, continued" +assert_equals "$(json_row keeper session)" "$KN" "the seat's alias names the job" +# One writer (a remote operator) that set a line for the old session, then for +# the job: the job's is the later, and clearing it leaves neither. +python3 - "$REPO_ROOT/skills/sminos/scripts" "$PF" "$KO" "$KN" <<'PY_CHARGES' +import sys +sys.path.insert(0, sys.argv[1]) +import sminos as mod +project = mod.find_project(sys.argv[2]) +for sid, text in ((sys.argv[3], "the old line"), (sys.argv[4], "the job's line")): + mod._append(project, mod.writer_id(), [{"kind": "in-charge", "session": sid, "text": text, "source": "set"}]) +PY_CHARGES +assert_equals "$(json_row keeper charge)" "the job's line" "lines one writer set along the chain are read in its order" +python3 - "$REPO_ROOT/skills/sminos/scripts" "$PF" "$KN" <<'PY_CLEAR' +import sys +sys.path.insert(0, sys.argv[1]) +import sminos as mod +mod._append(mod.find_project(sys.argv[2]), mod.writer_id(), + [{"kind": "in-charge", "session": sys.argv[3], "text": "", "source": "set"}]) +PY_CLEAR +assert_equals "$(json_row keeper charge_source)" "" "and a line it clears is cleared, the old one not read again" +rm -f "$HOME/.claude/sessions/pf-keeperjob.json" +"$SMINOS" remove pf-repo/keeper >/dev/null +# A seat resolved directly (no list between) when the hand-off is only now +# readable: the target is the job, by its alias and by the old session id +# alike, the id binding the job. +bound_pair() { # alias old new query — a seat whose hand-off is readable but not yet read + run "$SMINOS" seat add pf-repo "$1" --session "$2" + pf_peer "$1" "$2" idle user + run "$SMINOS" list + bg_peer "$1job" "$3" auto + run "$SMINOS" list + handoff_tx "$2" "$3" + rm -f "$HOME/.claude/sessions/pf-$1.json" +} +BO=8a00002c-0000-4000-8000-00000000008a +BN=8b00002d-0000-4000-8000-00000000008b +bound_pair binder "$BO" "$BN" +BOUND="$(python3 - "$REPO_ROOT/skills/sminos/scripts" "$BN" <<'PY_BOUND' +import sys +sys.path.insert(0, sys.argv[1]) +import sminos_bridge +print(sminos_bridge.watch_session("pf-repo/binder") == sys.argv[2]) +PY_BOUND +)" +assert_equals "$BOUND" "True" "a seat named by its alias, before any reading reconciled it, is watched as the job" +rm -f "$HOME/.claude/sessions/pf-binderjob.json" +"$SMINOS" remove pf-repo/binder >/dev/null +CO=7a00002a-0000-4000-8000-00000000007a +CN=7b00002b-0000-4000-8000-00000000007b +CP=7c00002e-0000-4000-8000-00000000007c +bound_pair clasp "$CO" "$CN" +run "$SMINOS" list +assert_equals "$(field "$CO" current)" "$CN" "a seat moves to the job it went on in" +bg_peer claspjob2 "$CP" auto +run "$SMINOS" list +handoff_tx "$CN" "$CP" # and the job is sent on in turn, unread yet +rm -f "$HOME/.claude/sessions/pf-claspjob.json" +BOUND="$(python3 - "$REPO_ROOT/skills/sminos/scripts" "$CN" "$CP" <<'PY_BOUND2' +import sys +sys.path.insert(0, sys.argv[1]) +import sminos +t = sminos.find_target(sys.argv[2]) +print(t[0] == "seat" and t[1]["current"] == sys.argv[3] and t[2] == sys.argv[3]) +PY_BOUND2 +)" +assert_equals "$BOUND" "True" "a name bound to the session a seat holds binds the job it went on in, once the seat is moved" +rm -f "$HOME/.claude/sessions/pf-claspjob2.json" +"$SMINOS" remove pf-repo/clasp >/dev/null +# A hand-off whose seat another command holds locked is left whole, not half +# done: nothing retired and the seat unmoved, and the next reading finishes it. +GO=6c000026-0000-4000-8000-00000000006c +GN=6d000027-0000-4000-8000-00000000006d +run "$SMINOS" seat add pf-repo gatekeeper --session "$GO" +pf_peer gatekeeper "$GO" idle user +run "$SMINOS" list +handoff_tx "$GO" "$GN" +rm -f "$HOME/.claude/sessions/pf-gatekeeper.json" "$TEST_ROOT/seat-held" "$TEST_ROOT/seat-release" +python3 - "$(lock_file pf-repo/gatekeeper)" "$TEST_ROOT/seat-held" "$TEST_ROOT/seat-release" <<'PY_SEATLOCK' & +import fcntl, os, sys, time +os.makedirs(os.path.dirname(sys.argv[1]), exist_ok=True) +with open(sys.argv[1], "a+") as f: + fcntl.flock(f, fcntl.LOCK_EX) + open(sys.argv[2], "w").close() + while not os.path.exists(sys.argv[3]): + time.sleep(0.05) +PY_SEATLOCK +SEATHOLDER=$! +for _ in $(seq 1 100); do [ -f "$TEST_ROOT/seat-held" ] && break; sleep .05; done +bg_peer gatekeeperjob "$GN" auto +run "$SMINOS" list +assert_not_contains "$(events_of "$GO")" "retired" "with the seat's lock held, the session it holds is not retired" +assert_equals "$(field "$GO" current)" "$GO" "nor the seat moved" +touch "$TEST_ROOT/seat-release"; wait "$SEATHOLDER" +run "$SMINOS" list +assert_equals "$(field "$GO" current)" "$GN" "once the lock is free, the next reading moves the seat" +assert_contains "$(events_of "$GO")" "retired:continued" "and retires the session it held, continued" +rm -f "$HOME/.claude/sessions/pf-gatekeeperjob.json" +"$SMINOS" remove pf-repo/gatekeeper >/dev/null +# A handed-off session the human resumes is its own member again: its id and +# short name it alone, and what it sets is not the job's. +VO=6e000028-0000-4000-8000-00000000006e +VN=6f000029-0000-4000-8000-00000000006f +pf_peer rover "$VO" idle derived +run "$SMINOS" list +handoff_tx "$VO" "$VN" +bg_peer roverjob "$VN" auto +run "$SMINOS" list +rm -f "$HOME/.claude/sessions/pf-rover.json" +run "$SMINOS" list +assert_contains "$(events_of "$VO")" "retired:continued" "a handed-off session is retired, continued" +printf '%s\n' '{"type":"user","message":{"role":"user","content":"back to it"}}' >> "$HOME/.claude/projects/pl-proj/$VO.jsonl" +printf '{"pid":%s,"sessionId":"%s","name":"rover-2a","nameSource":"derived","kind":"interactive","startedAt":%s,"status":"idle","cwd":"%s","messagingSocketPath":"%s"}\n' \ + "$SOCK_PID" "$VO" "$(python3 -c 'import time; print(int(time.time() * 1000))')" "$R" "$TEST_ROOT/in-rover.sock" \ + > "$HOME/.claude/sessions/pf-rover.json" +run "$SMINOS" list +assert_equals "$(events_of "$VO")" "joined retired:continued joined " "resumed, it re-joins" +session_row() { # session field → that field of the session's row in list --json + "$SMINOS" list --json | python3 -c 'import json, sys +rows = [d for d in json.load(sys.stdin) if d["session"] == sys.argv[1]] +v = rows[0].get(sys.argv[2]) if rows else "" +print(v if isinstance(v, str) else json.dumps(v))' "$1" "$2" +} +assert_equals "$(session_row "$VN" former)" "[]" "and the job no longer continues it" +run "$SMINOS" say "@${VO:0:8} to the resumed one" +assert_rc 0 "$RC" "its short names one member" +assert_contains "$(inbox rover)" "to the resumed one" "the resumed session" +assert_not_contains "$(inbox roverjob)" "to the resumed one" "not the job" +CLAUDE_CODE_SESSION_ID="$VO" run "$SMINOS" charge "the resumed line" +assert_rc 0 "$RC" "the resumed session sets its line" +assert_not_contains "$(session_row "$VN" charge)" "the resumed line" "which is not the job's" +rm -f "$HOME/.claude/sessions/pf-rover.json" "$HOME/.claude/sessions/pf-roverjob.json" +run "$SMINOS" list +# A name the harness made up is drawn again by every process: a resume +# keeps the member's alias. A name someone gives renames it. +RS=5d000021-0000-4000-8000-00000000005d +pf_peer drifter "$RS" idle derived +run "$SMINOS" list +rm -f "$HOME/.claude/sessions/pf-drifter.json" +run "$SMINOS" list +pf_peer drifter-3f "$RS" idle derived # resumed: the harness drew its name again +run "$SMINOS" list +assert_equals "$(events_of "$RS")" "joined retired:ended joined " "a resumed session re-joins" +assert_contains "$(tail -1 "$PFLOG/testhost.$RS.jsonl")" '"alias": "drifter"' "under the alias it held, not the name drawn again" +assert_equals "$(json_row drifter session)" "$RS" "and lists under it" +CLAUDE_CODE_SESSION_ID="$RS" run "$SMINOS" say "back again" +assert_contains "$(tail -1 "$PFLOG/testhost.$RS.jsonl")" '"from": "drifter"' "and speaks under it" +rm -f "$HOME/.claude/sessions/pf-drifter-3f.json" +pf_peer anchor "$RS" idle user +run "$SMINOS" list +assert_equals "$(json_row anchor session)" "$RS" "a name someone gives the session renames it" +rm -f "$HOME/.claude/sessions/pf-anchor.json" +run "$SMINOS" list + # A seat and a seat-less session that share an alias: every member verb # refuses the name as ambiguous, as a tag is refused, and #short picks one. DS=2f00000f-0000-4000-8000-00000000002f diff --git a/tests/sminos/test-bridge-control.sh b/tests/sminos/test-bridge-control.sh index a252cfcbe93..d4e0aac8b93 100755 --- a/tests/sminos/test-bridge-control.sh +++ b/tests/sminos/test-bridge-control.sh @@ -213,13 +213,15 @@ elif op == "fakemod": # sminos session lines-file mode-file result = {"ok": True, "result": {"queued": True, "turnId": "t1"}} elif o == "deliver": key = msg["key"] + # The mod's mark: per project, the bare writer for a key naming none. + mark = "%s/%s" % (key["project"], key["writer"]) if "project" in key else key["writer"] if mode == "refuse": - result = {"ok": False, "error": "deliver needs key {writer, seq} and text"} - elif key["seq"] <= delivered.get(key["writer"], 0): + result = {"ok": False, "error": "deliver needs key {project?, writer, seq} and text"} + elif key["seq"] <= delivered.get(mark, 0): events.append({"type": "delivered", "key": key, "outcome": "duplicate"}) result = {"ok": True, "result": {"duplicate": True}} else: - delivered[key["writer"]] = key["seq"] + delivered[mark] = key["seq"] events.append({"type": "delivered", "key": key, "outcome": "submitted"}) result = {"ok": True, "result": {"submitted": True}} else: @@ -414,6 +416,21 @@ for p in glob.glob(sys.argv[1] + "/projects/*/log/testhost.human.jsonl"): if e.get("text") == "@rover read the plan": print(e["seq"])' "$SMINOS_HOME")" assert_equals "$SEQ" "$LOGSEQ" "and its seq" +P1="$(field_of "$(cat "$R/.smns")" project)" +assert_equals "$(field_of "$L" key.project)" "$P1" "and its project's id" +# What the session kept holds the writer's bare mark, 3, from before keys +# named their project (#229): a project's own count starts under it. +assert_equals "$(( SEQ <= 3 ))" "1" "the event's seq is at or below the writer's bare mark the session kept" +assert_contains "$(cat "$KEPT")" "\"$P1/testhost/human\": $SEQ" "yet it was submitted: the child keeps the project's own mark" +assert_contains "$(cat "$KEPT")" '"testhost/human": 3' "beside the bare one, untouched" +assert_not_contains "$OUT" "already delivered" "and the sender was told of no duplicate" +# A duplicate is taken as delivered, but said, to the sender and in daemon.log. +set +e; OUT="$(PYTHONPATH="$REPO_ROOT/skills/sminos/scripts" python3 -c 'import sys, sminos_control as C +print(C.deliver(sys.argv[1], sys.argv[2], {"writer": "testhost/human", "seq": int(sys.argv[3])}, "again"))' "$SID" "$P1" "$SEQ" 2>&1)"; set -e +assert_contains "$OUT" "True" "a key the bridge already submitted is taken as delivered" +assert_contains "$OUT" "the bridge of session $SID took testhost/human:$SEQ as already delivered" "and said to the sender" +assert_contains "$(cat "$SMINOS_HOME/daemon.log" 2>/dev/null)" "the bridge of session $SID took testhost/human:$SEQ as already delivered" \ + "and in daemon.log, naming the event and the session" assert_contains "$(field_of "$L" text)" "[sminos chat ctldemo" "the text is the frame the inbox would carry" assert_contains "$(field_of "$L" text)" "read the plan" "with the message" assert_equals "$([ -s "$T/rover-inbox" ] && echo framed || echo none)" "none" "and nothing went to the inbox socket" @@ -421,7 +438,8 @@ set +e; OUT="$(cd "$R" && "$SMINOS" send rover "check the tests" 2>&1)"; RC=$?; assert_rc 0 "$RC" "send to a bridged member exits 0" L="$(last_op "$T/lines-b" deliver)" assert_contains "$(field_of "$L" text)" "[sminos message from human id=" "send's own frame went through the bridge" -assert_equals "$(( $(field_of "$L" key.seq) > SEQ ))" "1" "keyed by its own event" +SENDSEQ="$(field_of "$L" key.seq)" +assert_equals "$(( SENDSEQ > SEQ ))" "1" "keyed by its own event" # The bridge refusing a delivery (it did not submit) leaves the inbox frame. mode refuse set +e; OUT="$(cd "$R" && "$SMINOS" say "@rover second try" 2>&1)"; RC=$?; set -e @@ -429,6 +447,29 @@ assert_contains "$OUT" "rover: sent" "a delivery the bridge refuses still reache until_true 100 test -s "$T/rover-inbox" || true assert_contains "$(cat "$T/rover-inbox" 2>/dev/null)" "second try" "through its inbox socket" mode busy +# One writer delivering to one session from two projects (#229): every +# project numbers the writer's events on its own, so the second project's +# event is below what the first delivered, and arrives all the same. +R2="$T/repo2" +mkdir -p "$R2" +git -c init.defaultBranch=main init -q "$R2" +(cd "$R2" && "$SMINOS" project init --name ctlsecond >/dev/null) +P2="$(field_of "$(cat "$R2/.smns")" project)" +printf '{"pid":%s,"sessionId":"%s","name":"rover","nameSource":"user","kind":"interactive","status":"idle","cwd":"%s","messagingSocketPath":"%s"}\n' \ + "$$" "$SID" "$R2" "$INBOX" > "$HOME/.claude/sessions/rover.json" +: > "$T/rover-inbox" +set +e; OUT="$(cd "$R2" && "$SMINOS" say "@rover from the second project" 2>&1)"; RC=$?; set -e +assert_rc 0 "$RC" "a say to the same member from a second project exits 0" +L="$(last_op "$T/lines-b" deliver)" +assert_contains "$(field_of "$L" text)" "from the second project" "it went through the bridge" +assert_equals "$(field_of "$L" key.project)" "$P2" "keyed by the second project" +assert_equals "$(( $(field_of "$L" key.seq) < SENDSEQ ))" "1" "its seq below what the first project delivered from the writer" +assert_equals "$(field_of "$(cat "$KEPT")" "delivered.$P2/testhost/human")" "$(field_of "$L" key.seq)" \ + "and it was submitted: the child keeps the second project's mark" +assert_not_contains "$OUT" "already delivered" "with no duplicate said" +assert_equals "$([ -s "$T/rover-inbox" ] && echo framed || echo none)" "none" "and nothing went to the inbox socket" +printf '{"pid":%s,"sessionId":"%s","name":"rover","nameSource":"user","kind":"interactive","status":"idle","cwd":"%s","messagingSocketPath":"%s"}\n' \ + "$$" "$SID" "$R" "$INBOX" > "$HOME/.claude/sessions/rover.json" # A --bg job with no bridge: a human's event goes through the engine # supervisor's control socket (`reply`, human-origin), else the inbox frame. diff --git a/tests/sminos/test_policy_requests.py b/tests/sminos/test_policy_requests.py index 3ecce0938f3..7de955cb20e 100644 --- a/tests/sminos/test_policy_requests.py +++ b/tests/sminos/test_policy_requests.py @@ -688,7 +688,7 @@ def test_b_a_web_send_dropped_after_submitted_is_reported_failed_once(self): self.deliver() self.assertIn("dropped after it was taken", self.bridge.texts()) self.assertEqual([o for o, _ in self.outcomes_of(e)], ["delivered"]) - key = {"writer": e["writer"], "seq": e["seq"]} + key = {"project": self.pid, "writer": e["writer"], "seq": e["seq"]} self.assertTrue(wait_for(lambda: [r for r in self.kept_refused() if r["key"] == key]), "the child kept the refusal") self.a_pass() @@ -709,7 +709,7 @@ def test_b2_a_refusal_whose_post_was_lost_is_reported_from_the_resync(self): e = self.human_from_hub("dropped, and its refusal's post lost") self.deliver() self.assertIn("dropped, and its refusal's post lost", self.bridge.texts()) - key = {"writer": e["writer"], "seq": e["seq"]} + key = {"project": self.pid, "writer": e["writer"], "seq": e["seq"]} self.assertTrue(wait_for(lambda: [r for r in self.kept_refused() if r["key"] == key]), "the child kept the refusal from the resync") self.bridge.post({"type": "resync", "turns": [], "streaming": [], "agents": [], "override": None, @@ -732,7 +732,7 @@ def test_c_a_refusal_of_an_event_written_here_is_not_reported(self): **{"from": "human", "to": [{"session": self.sid, "alias": self.alias}], "text": "written here, dropped"}) self.deliver() - key = {"writer": e["writer"], "seq": e["seq"]} + key = {"project": self.pid, "writer": e["writer"], "seq": e["seq"]} self.assertTrue(wait_for(lambda: [r for r in self.kept_refused() if r["key"] == key])) self.a_pass() self.assertEqual(self.outcomes_of(e), [], "nothing from outside, nothing reported") @@ -744,7 +744,7 @@ def test_e_a_marker_that_cannot_be_written_is_said_once_and_held_by_the_process( self.bridge.refusing = "dropped, marker unwritable" e = self.human_from_hub("its marker will not be kept") self.deliver() - key = {"writer": e["writer"], "seq": e["seq"]} + key = {"project": self.pid, "writer": e["writer"], "seq": e["seq"]} self.assertTrue(wait_for(lambda: [r for r in self.kept_refused() if r["key"] == key])) real = sminos.write_json @@ -791,6 +791,30 @@ def test_d_a_refusal_is_reported_only_once_its_first_outcome_is_recorded(self): except OSError: pass + def test_f_a_refusal_naming_another_project_is_not_reported_here(self): + """Each project numbers a writer's events on its own: a refusal the + bridge kept for another project's event of the same writer and seq + is that project's, and reports nothing for this one's.""" + import sminos_bridge as bridge + other = Session(self.root) + try: + sminos.reconcile(self.project) + alias = member_of(self.project, other.sid).alias + e = self.human_from_hub("delivered, refused only elsewhere", other.sid, alias) + key = {"project": "elsewhere", "writer": e["writer"], "seq": e["seq"]} + with open(bridge.kept_path(other.sid), "w") as f: + json.dump({"override": None, "delivered": {}, "refused": [{"key": key, "reason": "dropped"}]}, f) + self.a_pass() + self.assertIn("delivered, refused only elsewhere", other.inbox.texts()) + self.a_pass() + self.assertEqual([o for o, _ in self.outcomes_of(e, other.sid)], ["delivered"]) + finally: + other.end() + try: + os.unlink(bridge.kept_path(other.sid)) + except OSError: + pass + class Requests(unittest.TestCase): """The relay of requests, against the real bridge child.""" @@ -1221,7 +1245,8 @@ def test_e_a_send_the_engine_dropped_after_submitted_reaches_the_hub_failed(self ofkey = lambda ms: [m["body"] for m in self.mine(ms[count:], "delivery") if m["body"]["event"] == key] # noqa: E731 self.assertIsNotNone(self.hub.wait(ofkey, timeout=15), "delivered first") self.assertTrue(wait_for(lambda: [r for r in bridge.load_kept(self.sid).get("refused") or [] - if r["key"] == key]), "the child kept the refusal") + if r["key"] == {"project": self.project, **key}]), + "the child kept the refusal, its key naming the project") r = cli("daemon", "tick") self.assertEqual(r.returncode, 0, r.stderr) ms = self.hub.wait(lambda ms: [d for d in ofkey(ms) if d["outcome"] == "failed"], timeout=15)