ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-12 23:12:23 -06:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
turnstone/tests/test_session_manager.py
T
Patrick Buckley 480a1426b3 Fail-closed history-commit handoff (#1005) 
* fix(session): fail-closed history-commit handoff (#981)

The deleted-workstream discovery is now a terminal, ws_id-keyed latch:
keyed conversation commits refuse admission once the durable parent is
gone (convergence finalizers and force-abandon are exempt), history
handoff refuses to mint a proof token so /history fails closed with a
503 instead of silently wiping the pane, and the SSE stream carries a
workstream_gone resync reason. Discarded commits leave a forensic log
of commit keys and roles, never content.

Conversation rows gain a commit_key (migration 071): keyed saves are
idempotent under retry, validated against the full commit identity, and
refused when they would cross a workstream deletion. The prune orphan
category now requires a NULL alias plus a two-hour updated grace, with
cutoffs computed at discovery time and carried into both dialects'
rechecks.

The mid-turn interjection queue is owner-partitioned with no per-site
mode flags: pops take the acting principal's and unowned rows, other
participants' rows are structurally retained, and enforcement lives at
queue admission plus the shared before_spawn gates. The retraction
ledger is bounded by open pop windows: pops open a window atomically
with the queue delete, restores close their ids atomically with the
ledger consume, every other exit closes through one helper, and misses
for unheld ids record nothing. The workstream-gone latch refuses
unattended wakes at all three gates (watcher spawn, claim, delivery
pre-pop), and the retry dispatcher regained its pre-envelope
cancel/error convergence net.

Persistence-state reporting derives through the session bound to each
UI instead of a registry lookup by id that failed open to healthy
during tombstone retention. The dashboard roster no longer re-inserts
ghost entries from trailing activity events, the history tool-outcome
scan tolerates interleaved non-turn rows, and the shared
handoff-deadline handle owns its own retirement.

Single-sourced across call sites: keyed-commit row values, attachment
save wrappers, tail-truncation and conflict-resolution bodies for both
storage dialects; worker-slot lifecycle field sets; the direct-commit
admission frame; queued-row layout accessors; the string-aware comment
stripper shared by every JS harness suite.

Refs #981 #964

* fix(session): sweep handoff fixes to their sibling surfaces

The interactive replay loop treated a system row as a tool-batch
boundary, so every tool result after an interleaved row vanished from
that pane while the coordinator rendered the same history correctly.
Only a conversational turn ends the batch window now, matching the
shared outcome index.

Accepted user turns clear the composer's attachment chips on the same
viewer policy that settles optimistic bubbles rather than on having
matched a local bubble, so a workstream created with an upload no
longer keeps a chip for an attachment the create dispatch already
consumed. The coordinator's raced-Stop arm emits the stream-end hook it
inherits alongside the idle state, leaving no unfinalized bubble or
unflushed tool output. Ending a session surfaces a failure toast when
the request never lands or answers with a non-JSON body.

The per-second persistence reconcile now probes each session without
blocking: a workstream whose generation and handoff locks are held is
skipped until the next pass instead of contending the locks every
commit needs. The one-shot repair that gates workstream creation at
capacity keeps a definite probe — it has no next pass, and the sessions
likeliest to be contended are the ones whose unresolved journals
emptied its candidate list.

Single-sourced: the attachment lane builds its conversation row through
the shared commit-identity builder; the ordinary worker exit releases
its slot through the lifecycle owner; both operator surfaces snapshot
their counters through one non-consuming helper; the replay preamble
loses its per-kind wrappers and its config hook; the browser harness
suites share one brace walker; and each in-flight history attempt is
one record carrying both its abort controller and its deadline.

Refs #981 #964
2026-08-11 04:18:36 -07:00

2609 lines
94 KiB
Python
Raw Permalink Blame History

"""Smoke tests for the unified SessionManager.
Cover the concurrency-sensitive paths that historically diverged
between ``WorkstreamManager`` and ``CoordinatorManager`` — slot
reservation, eviction, per-ws lock serialization on lazy rehydrate,
state flips, close-unblocks-UI. All exercised through a
``FakeAdapter`` that records ``emit_*`` / ``cleanup_ui`` calls so
tests can assert the transport contract without spinning up real
WebUI / ClusterCollector pipelines. The adapter is wired as both the
manager's ``adapter`` (for ``cleanup_ui`` / ``build_*``) and its
``event_emitter`` (for the ``emit_*`` lifecycle calls); production
adapters do the same — interactive on ``server.py``, coordinator on
``console/server.py``.
"""
from __future__ import annotations
import threading
import time
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
from unittest.mock import MagicMock
import pytest
from turnstone.core.model_registry import ModelClientConstructionError, UnknownModelAliasError
from turnstone.core.session_manager import SessionKindAdapter, SessionManager
from turnstone.core.workstream import (
BULK_CLOSE_STATE_VALUES,
Workstream,
WorkstreamKind,
WorkstreamState,
concrete_method,
)
# ---------------------------------------------------------------------------
# Test fixtures
# ---------------------------------------------------------------------------
@dataclass
class _Event:
kind: str # "created" | "rehydrated" | "state" | "closed"
ws_id: str
state: WorkstreamState | None = None
reason: str | None = None
name: str | None = None
class FakeUI:
"""Minimal UI stand-in with the events close() needs to unblock."""
def __init__(self) -> None:
self.approval_unblocked = False
self.plan_unblocked = False
self.fg_unblocked = False
self.closed_broadcast = False
def _unblock(self) -> None:
self.approval_unblocked = True
self.plan_unblocked = True
self.fg_unblocked = True
def broadcast_ws_closed(self) -> None:
self.closed_broadcast = True
class FakeSession:
"""Minimal ChatSession stand-in; exposes cancel / close / resume."""
def __init__(self, ws_id: str, *, model_alias: str | None = None) -> None:
self.ws_id = ws_id
self.model_alias = model_alias
self.cancelled = False
self.closed = False
self.resumed = False
self.resume_hook: Callable[[], None] | None = None
def cancel(self) -> None:
self.cancelled = True
def close(self) -> None:
self.closed = True
def resume(self, ws_id: str) -> None:
if self.resume_hook is not None:
self.resume_hook()
self.resumed = True
class FakeAdapter:
"""Records events + builds FakeUI / FakeSession for tests."""
def __init__(
self,
kind: WorkstreamKind = WorkstreamKind.INTERACTIVE,
*,
build_session_raises: bool = False,
) -> None:
self.kind = kind
self.events: list[_Event] = []
self._events_lock = threading.Lock()
self.cleaned_up: list[str] = []
self.build_session_calls = 0
self.build_session_raises = build_session_raises
self.last_build_model: object | None = None
self.build_models: list[object | None] = []
self.built_sessions: list[FakeSession] = []
self.build_session_hook: Callable[[Workstream, object | None], FakeSession] | None = None
# Slow down session build so concurrent tests can race.
self.build_session_delay = 0.0
def emit_created(self, ws: Workstream) -> None:
with self._events_lock:
self.events.append(_Event("created", ws.id))
def emit_rehydrated(self, ws: Workstream) -> None:
# Distinct from "created" so a regression where the manager
# fires emit_created on the rehydrate path (or emit_rehydrated
# on the create path) actually fails a test. The
# SessionEventEmitter Protocol carves these out as semantically
# different — coord uses emit_rehydrated for the storage-seeded
# subtree rebuild that emit_created skips.
with self._events_lock:
self.events.append(_Event("rehydrated", ws.id))
def emit_state(self, ws: Workstream, state: WorkstreamState) -> None:
with self._events_lock:
self.events.append(_Event("state", ws.id, state=state))
def emit_closed(
self,
ws_id: str,
*,
reason: str = "closed",
name: str = "",
) -> None:
with self._events_lock:
self.events.append(_Event("closed", ws_id, reason=reason, name=name))
def cleanup_ui(self, ws: Workstream) -> None:
self.cleaned_up.append(ws.id)
if ws.session is not None:
ws.session.cancel()
ws.session.close()
if ws.ui is not None:
ws.ui._unblock()
ws.ui.broadcast_ws_closed()
def build_ui(self, ws: Workstream) -> Any:
return FakeUI()
def build_session(self, ws: Workstream, **kwargs: object) -> Any:
self.build_session_calls += 1
# Record the ``model`` kwarg (None on fresh-create, the saved
# alias on rehydrate) so tests can assert SessionManager.open()
# threads the persisted alias through to construction instead
# of letting the adapter resolve the *current* default alias.
model = kwargs.get("model")
self.last_build_model = model
self.build_models.append(model)
if self.build_session_delay:
time.sleep(self.build_session_delay)
if self.build_session_raises:
raise RuntimeError("build_session forced failure")
session = (
self.build_session_hook(ws, model)
if self.build_session_hook is not None
else FakeSession(ws.id)
)
self.built_sessions.append(session)
return session
def events_of(self, kind: str) -> list[_Event]:
with self._events_lock:
return [e for e in self.events if e.kind == kind]
class _FakeRowMapping:
"""SQLAlchemy-Row-like wrapper exposing ``_mapping`` over a ``_Row``.
The real backends return ``Row`` objects with a ``_mapping`` attribute;
consumers (e.g. ``CoordinatorIdleObserver._active_children``) prefer
``row._mapping[<col>]`` access. This shim mirrors that contract so
fakes are interchangeable with real Rows in tests.
"""
def __init__(self, row: _Row) -> None:
self._mapping = {
"ws_id": row.ws_id,
"user_id": row.user_id,
"name": row.name,
"kind": row.kind,
"state": row.state,
"parent_ws_id": row.parent_ws_id,
"updated": row.updated,
"node_id": row.node_id,
}
@dataclass
class _Row:
ws_id: str
user_id: str
name: str
kind: str
state: str = "idle"
parent_ws_id: str | None = None
updated: str = ""
node_id: str | None = None
project_id: str | None = None
persona: str | None = None
class FakeStorage:
"""In-memory storage that mirrors the StorageBackend surface the manager uses."""
def __init__(self) -> None:
self.rows: dict[str, _Row] = {}
self.state_updates: list[tuple[str, str]] = []
self.touch_calls: list[str] = []
self.register_raises = False
self.lock = threading.Lock()
# Live-services lookup target for close_idle pass 2. Map
# service_type → list of live service_ids. Tests that exercise
# liveness scoping populate this directly; default empty means
# "no peers alive" (every row unprotected by liveness).
self.live_services: dict[str, list[str]] = {}
self.list_services_raises = False
self.delete_stale_creating_raises = False
# Per-ws config (model_alias, temperature, …). Populated by
# tests that exercise the rehydrate-preserves-config path; the
# SessionManager.open() rehydrate path reads this through
# ``self._storage.load_workstream_config`` so it can pass the
# saved alias into ``build_session`` and avoid clobbering the
# original on construction.
self.ws_config: dict[str, dict[str, str]] = {}
self.fork_reservations: dict[str, str] = {}
@staticmethod
def _now_iso() -> str:
return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
def register_workstream(
self,
ws_id: str,
*,
node_id: str | None = None,
user_id: str | None = None,
name: str = "",
kind: WorkstreamKind | str = WorkstreamKind.INTERACTIVE,
parent_ws_id: str | None = None,
project_id: str | None = None,
persona: str | None = None,
skill_id: str = "",
skill_version: int = 0,
state: str = "idle",
updated: str | None = None,
fork_reservation_token: str = "",
) -> None:
if self.register_raises:
raise RuntimeError("register forced failure")
kind_str = kind.value if isinstance(kind, WorkstreamKind) else str(kind)
with self.lock:
self.rows[ws_id] = _Row(
ws_id=ws_id,
user_id=user_id or "",
name=name,
kind=kind_str,
state=state,
parent_ws_id=parent_ws_id,
updated=updated if updated is not None else self._now_iso(),
node_id=node_id,
project_id=project_id,
persona=persona if persona else None,
)
if fork_reservation_token:
self.fork_reservations[ws_id] = fork_reservation_token
def touch_workstream(self, ws_id: str) -> None:
with self.lock:
self.touch_calls.append(ws_id)
if ws_id in self.rows:
self.rows[ws_id].updated = self._now_iso()
def update_workstream_state(self, ws_id: str, state: str) -> None:
with self.lock:
self.state_updates.append((ws_id, state))
if ws_id in self.rows:
self.rows[ws_id].state = state
self.rows[ws_id].updated = self._now_iso()
def bulk_close_stale_orphans(
self,
kind: WorkstreamKind | str,
cutoff: str,
exclude_ws_ids: list[str],
live_node_ids: list[str] | None = None,
) -> list[str]:
kind_str = kind.value if isinstance(kind, WorkstreamKind) else str(kind)
excluded = set(exclude_ws_ids)
live_set = set(live_node_ids) if live_node_ids else set()
now = self._now_iso()
closed: list[str] = []
with self.lock:
for ws_id, row in self.rows.items():
if (
row.kind == kind_str
and row.state in BULK_CLOSE_STATE_VALUES
and row.updated < cutoff
and ws_id not in excluded
):
# Liveness gate: when live_node_ids was provided AND
# non-empty, protect rows whose owner is in the live
# set. NULL node_id is always eligible. When
# live_node_ids is None or empty, no protection
# (mirror of the real backends).
if live_node_ids and row.node_id is not None and row.node_id in live_set:
continue
row.state = "closed"
row.updated = now
self.state_updates.append((ws_id, "closed"))
closed.append(ws_id)
return closed
def delete_stale_creating_reservations(
self,
kind: WorkstreamKind | str,
cutoff: str,
exclude_ws_ids: list[str],
*,
live_node_ids: list[str],
local_node_id: str | None,
) -> list[str]:
if live_node_ids is None: # type: ignore[comparison-overlap]
return []
if self.delete_stale_creating_raises:
raise RuntimeError("stale creating delete forced failure")
kind_str = kind.value if isinstance(kind, WorkstreamKind) else str(kind)
excluded = set(exclude_ws_ids)
protected_live = {
node_id for node_id in live_node_ids if node_id and node_id != local_node_id
}
deleted: list[str] = []
with self.lock:
for ws_id, row in list(self.rows.items()):
if (
row.kind != kind_str
or row.state != "creating"
or row.updated >= cutoff
or ws_id in excluded
):
continue
if row.node_id is not None and row.node_id in protected_live:
continue
self.rows.pop(ws_id, None)
self.ws_config.pop(ws_id, None)
self.fork_reservations.pop(ws_id, None)
deleted.append(ws_id)
return deleted
def list_services(self, service_type: str, max_age_seconds: int = 120) -> list[dict[str, str]]:
if self.list_services_raises:
raise RuntimeError("list_services forced failure")
with self.lock:
return [
{"service_id": sid, "service_type": service_type}
for sid in self.live_services.get(service_type, [])
]
def get_workstream(self, ws_id: str) -> dict[str, Any] | None:
with self.lock:
row = self.rows.get(ws_id)
if row is None:
return None
return {
"ws_id": row.ws_id,
"user_id": row.user_id,
"name": row.name,
"kind": row.kind,
"state": row.state,
"parent_ws_id": row.parent_ws_id,
"persona": row.persona,
}
def ensure_workstream_incarnation_snapshot(self, ws_id: str) -> dict[str, Any] | None:
with self.lock:
row = self.rows.get(ws_id)
if row is None:
return None
token = self.fork_reservations.get(ws_id)
if not token:
token = uuid.uuid4().hex
self.fork_reservations[ws_id] = token
return {
"ws_id": row.ws_id,
"user_id": row.user_id,
"name": row.name,
"kind": row.kind,
"state": row.state,
"parent_ws_id": row.parent_ws_id,
"project_id": row.project_id,
"persona": row.persona,
"fork_reservation_token": token,
}
def list_workstreams(
self,
node_id: str | None = None,
limit: int = 100,
*,
parent_ws_id: str | None = None,
kind: WorkstreamKind | str | None = None,
user_id: str | None = None,
) -> list[Any]:
kind_str = kind.value if isinstance(kind, WorkstreamKind) else kind
with self.lock:
matched: list[_FakeRowMapping] = []
for row in self.rows.values():
if parent_ws_id is not None and row.parent_ws_id != parent_ws_id:
continue
if kind_str is not None and row.kind != kind_str:
continue
if user_id is not None and row.user_id != user_id:
continue
matched.append(_FakeRowMapping(row))
# Order by updated DESC so the consumer's LIMIT semantics match
# production (storage backends order this way).
matched.sort(key=lambda r: r._mapping["updated"], reverse=True)
return matched[:limit]
def count_workstreams_by_state(
self,
*,
parent_ws_id: str | None = None,
user_id: str | None = None,
) -> dict[str, int]:
counts: dict[str, int] = {}
with self.lock:
for row in self.rows.values():
if parent_ws_id is not None and row.parent_ws_id != parent_ws_id:
continue
if user_id is not None and row.user_id != user_id:
continue
counts[row.state] = counts.get(row.state, 0) + 1
return counts
def delete_workstream(self, ws_id: str) -> None:
with self.lock:
self.rows.pop(ws_id, None)
self.ws_config.pop(ws_id, None)
self.fork_reservations.pop(ws_id, None)
def delete_workstream_if_fork_reserved(
self,
ws_id: str,
fork_reservation_token: str,
) -> bool:
with self.lock:
if self.fork_reservations.get(ws_id) != fork_reservation_token:
return False
self.rows.pop(ws_id, None)
self.ws_config.pop(ws_id, None)
self.fork_reservations.pop(ws_id, None)
return True
def publish_deferred_create(
self,
ws_id: str,
fork_reservation_token: str,
) -> bool:
with self.lock:
row = self.rows.get(ws_id)
if (
row is None
or row.state != "creating"
or self.fork_reservations.get(ws_id) != fork_reservation_token
):
return False
row.state = "idle"
row.updated = self._now_iso()
return True
def get_workstream_reservation_token(self, ws_id: str) -> str:
with self.lock:
return self.fork_reservations.get(ws_id, "")
def count_skill_versions(self, template_id: str) -> int:
return 0
def load_workstream_config(self, ws_id: str) -> dict[str, str]:
with self.lock:
return dict(self.ws_config.get(ws_id, {}))
def save_workstream_config(self, ws_id: str, config: dict[str, str]) -> None:
# Mirrors the real backend's INSERT OR REPLACE per-key semantics
# — callers expect a partial save to overwrite only the keys
# they pass, not the whole row.
with self.lock:
row = self.ws_config.setdefault(ws_id, {})
row.update(config)
_EMITTER_DEFAULT = object()
def _make_manager(
adapter: FakeAdapter | None = None,
*,
max_active: int = 5,
storage: FakeStorage | None = None,
event_emitter: Any = _EMITTER_DEFAULT,
node_id: str | None = None,
model_validator: Callable[[str], bool] | None = None,
) -> tuple[SessionManager, FakeAdapter, FakeStorage]:
"""Build a SessionManager wired to a FakeAdapter for both Protocols.
``event_emitter`` defaults to the adapter (production wiring shape);
pass ``None`` explicitly to disable the lifecycle-event side
channel for tests that care about no-emitter behaviour.
"""
adapter = adapter or FakeAdapter()
storage = storage or FakeStorage()
# FakeAdapter implements both Protocols (the production adapters
# do too — wire as both so the emit_* assertions in this file
# still see the events the manager fires).
emitter = adapter if event_emitter is _EMITTER_DEFAULT else event_emitter
mgr = SessionManager(
adapter,
storage=storage,
max_active=max_active,
event_emitter=emitter,
node_id=node_id,
model_validator=model_validator,
)
return mgr, adapter, storage
# ---------------------------------------------------------------------------
# Construction
# ---------------------------------------------------------------------------
def test_session_manager_constructs_with_adapter() -> None:
mgr, adapter, _ = _make_manager()
assert mgr.max_active == 5
assert mgr.kind == adapter.kind
def test_session_manager_rejects_invalid_max_active() -> None:
with pytest.raises(ValueError, match="max_active must be >= 1"):
SessionManager(FakeAdapter(), storage=FakeStorage(), max_active=0)
def test_noop_adapter_satisfies_protocol() -> None:
adapter: SessionKindAdapter = FakeAdapter()
assert adapter.kind == WorkstreamKind.INTERACTIVE
# ---------------------------------------------------------------------------
# create
# ---------------------------------------------------------------------------
def test_create_persists_and_emits_created() -> None:
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1", name="hello")
assert ws.user_id == "u1"
assert ws.name == "hello"
assert ws.session is not None
assert ws.ui is not None
assert ws.id in storage.rows
assert storage.rows[ws.id].kind == WorkstreamKind.INTERACTIVE.value
assert [e.ws_id for e in adapter.events_of("created")] == [ws.id]
def test_create_with_defer_emit_created_skips_emit() -> None:
"""``defer_emit_created=True`` returns the workstream but skips
the ``emit_created`` call. The slot, storage row, and built
session all exist — only the lifecycle event is held back.
Caller takes ownership of advertising the workstream via
:meth:`SessionManager.commit_create` (success) or
:meth:`SessionManager.discard` (rollback)."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1", name="deferred", defer_emit_created=True)
# Workstream is fully constructed, but stays hidden from ordinary manager
# lookup until its lifecycle birth is committed.
assert ws.session is not None
assert ws.ui is not None
assert ws.id in storage.rows
assert mgr.get(ws.id) is None
# No created event fired.
assert adapter.events_of("created") == []
def test_commit_create_fires_deferred_emit_created() -> None:
"""``commit_create`` is the deferred counterpart that fires the
pending ``emit_created`` event after the caller's post-create
work (e.g. attachment validation in the lifted HTTP handler)
confirms the workstream should be advertised."""
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", defer_emit_created=True)
assert adapter.events_of("created") == []
mgr.commit_create(ws)
assert [e.ws_id for e in adapter.events_of("created")] == [ws.id]
assert mgr.get(ws.id) is ws
def test_commit_create_is_noop_without_event_emitter() -> None:
"""``commit_create`` must tolerate a manager constructed without
an event emitter — the deferred-create + commit pair has to
work the same shape regardless of whether transport fan-out is
wired (test fixtures, future kinds without an emitter)."""
mgr, adapter, _ = _make_manager(event_emitter=None)
ws = mgr.create(user_id="u1", defer_emit_created=True)
# Must not raise; nothing observable should happen.
mgr.commit_create(ws)
assert adapter.events_of("created") == []
def test_discard_releases_slot_without_emit_closed() -> None:
"""``discard`` is the rollback counterpart to ``commit_create``.
Releases the in-memory slot + cleans up the UI, but does NOT
fire ``emit_closed`` because the workstream's existence was
never advertised (caller used ``defer_emit_created=True``).
Distinct from ``close`` which DOES fire ``emit_closed`` to
advertise the transition. Pre-fix, the lifted create handler's
rollback path called ``close`` and produced a phantom
create→close pair on the cluster events stream when attachment
validation failed; ``discard`` is the surgical fix."""
mgr, adapter, storage = _make_manager(max_active=2)
ws = mgr.create(user_id="u1", name="will-be-discarded", defer_emit_created=True)
ws_id = ws.id
discarded = mgr.discard(ws_id)
assert discarded is True
# In-memory slot released — capacity restored.
assert mgr.get(ws_id) is None
assert mgr.count == 0
# UI cleanup ran (cleanup_ui is part of discard's contract).
assert ws_id in adapter.cleaned_up
# No advertisement at any point: no created event, no closed event.
assert adapter.events_of("created") == []
assert adapter.events_of("closed") == []
# Storage row survives — caller is responsible for
# ``delete_workstream`` if they want a complete rollback. Mirrors
# mgr.create's own session-build-failure path.
assert ws_id in storage.rows
def test_discard_returns_false_for_unknown_ws_id() -> None:
"""``discard`` is idempotent on an absent ws_id — returns False
instead of raising so the lifted handler's rollback path can
safely call discard inside ``contextlib.suppress`` without
spurious failures swallowing real errors."""
mgr, _, _ = _make_manager()
result = mgr.discard("nonexistent-ws-id")
assert result is False
def test_commit_create_after_discard_is_no_op(caplog) -> None:
"""Caller-bug case: ``commit_create`` after ``discard`` must not
fire ``emit_created`` for a workstream that's no longer tracked
by the manager. The workstream object still exists in the
caller's scope (discard only removed it from the manager's slot
map), so forwarding it to ``commit_create`` would fire a phantom
``ws_created`` for an id the cluster collector / children
registry will then never see ``ws_closed`` for. The guard logs
``session_mgr.commit_create.untracked`` and returns without
emitting."""
import logging
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", defer_emit_created=True)
discarded = mgr.discard(ws.id)
assert discarded is True
assert adapter.events_of("created") == []
with caplog.at_level(logging.WARNING, logger="turnstone.core.session_manager"):
mgr.commit_create(ws)
# No event emitted — the tracked-ws check failed.
assert adapter.events_of("created") == []
# Warning surfaced for operator triage.
assert any("commit_create.untracked" in record.message for record in caplog.records), [
r.message for r in caplog.records
]
def test_commit_create_is_idempotent_on_duplicate_call(caplog) -> None:
"""Caller-bug case: calling ``commit_create`` twice on the same
workstream must fire ``emit_created`` exactly once. The second
call hits the ``_emit_created_fired`` guard, logs
``session_mgr.commit_create.already_fired``, and returns without
re-emitting."""
import logging
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", defer_emit_created=True)
mgr.commit_create(ws)
assert [e.ws_id for e in adapter.events_of("created")] == [ws.id]
with caplog.at_level(logging.WARNING, logger="turnstone.core.session_manager"):
mgr.commit_create(ws)
# Still exactly one created event — the guard short-circuited
# the second call.
assert [e.ws_id for e in adapter.events_of("created")] == [ws.id]
assert any("commit_create.already_fired" in record.message for record in caplog.records), [
r.message for r in caplog.records
]
def test_discard_after_emit_created_warns_but_releases_slot(caplog) -> None:
"""Caller-bug case: ``discard`` on a workstream where
``emit_created`` has already fired (either via the non-deferred
create path or via ``commit_create``). The intended retraction
path is ``close`` (which fires ``emit_closed``); ``discard``
leaves a stale ``ws_created`` on the wire with no matching
``ws_closed``, which is exactly the phantom-event bug
``defer_emit_created`` was added to fix.
``discard`` still releases the in-memory slot (so capacity is
freed even when the caller misuses the API) but logs a
``warning`` so the misuse surfaces in ops logs.
"""
import logging
mgr, adapter, _ = _make_manager()
# Non-deferred create — emit_created fires inside mgr.create.
ws = mgr.create(user_id="u1", name="created-then-discarded")
assert [e.ws_id for e in adapter.events_of("created")] == [ws.id]
with caplog.at_level(logging.WARNING, logger="turnstone.core.session_manager"):
result = mgr.discard(ws.id)
# Slot released (caller-bug doesn't strand capacity).
assert result is True
assert mgr.get(ws.id) is None
# Warning surfaced for operator triage.
assert any("discard.after_emit_created" in record.message for record in caplog.records), [
r.message for r in caplog.records
]
# No ws_closed was fired — the bug being warned about is exactly
# this asymmetry (created without close on the wire). Operators
# who actually want a clean retraction should call close instead.
assert adapter.events_of("closed") == []
def test_create_evicts_oldest_idle_at_capacity() -> None:
mgr, adapter, _ = _make_manager(max_active=2)
first = mgr.create(user_id="u1")
# Nudge the timestamp so 'first' is the clear eviction candidate.
first.last_active = time.monotonic() - 100
mgr.create(user_id="u1")
# Third create triggers eviction of the oldest IDLE (= first).
third = mgr.create(user_id="u1")
assert mgr.get(first.id) is None
assert mgr.get(third.id) is not None
# Adapter transport saw the eviction and the new create.
assert first.id in [e.ws_id for e in adapter.events_of("closed")]
assert third.id in [e.ws_id for e in adapter.events_of("created")]
assert first.id in adapter.cleaned_up
def test_create_raises_when_all_active_and_no_idle() -> None:
mgr, _, _ = _make_manager(max_active=1)
ws = mgr.create(user_id="u1")
ws.state = WorkstreamState.RUNNING # block eviction
with pytest.raises(RuntimeError, match="slots are active"):
mgr.create(user_id="u1")
def test_create_rolls_back_slot_on_session_failure() -> None:
adapter = FakeAdapter(build_session_raises=True)
mgr, _, storage = _make_manager(adapter=adapter)
with pytest.raises(RuntimeError, match="build_session forced failure"):
mgr.create(user_id="u1")
# Slot and hidden durable reservation are both released. A ``creating``
# row is intentionally undiscoverable, so retaining it would leak an
# unrecoverable workstream and its incarnation token.
assert mgr.count == 0
assert storage.rows == {}
def test_failed_pending_fork_create_deletes_exact_storage_reservation() -> None:
adapter = FakeAdapter(build_session_raises=True)
mgr, _, storage = _make_manager(adapter=adapter)
with pytest.raises(RuntimeError, match="build_session forced failure"):
mgr.create(
user_id="u1",
defer_emit_created=True,
_fork_reservation=True,
)
assert mgr.count == 0
assert storage.rows == {}
assert storage.fork_reservations == {}
def test_create_rolls_back_slot_on_persist_failure() -> None:
storage = FakeStorage()
storage.register_raises = True
mgr, _, _ = _make_manager(storage=storage)
with pytest.raises(RuntimeError, match="register forced failure"):
mgr.create(user_id="u1")
assert mgr.count == 0
def test_concurrent_create_does_not_exceed_max_active() -> None:
mgr, _, _ = _make_manager(max_active=3)
adapter = mgr._adapter # type: ignore[attr-defined]
assert isinstance(adapter, FakeAdapter)
adapter.build_session_delay = 0.02 # widen the race window
results: list[Workstream | Exception] = []
lock = threading.Lock()
def _create() -> None:
try:
ws = mgr.create(user_id="u1")
with lock:
results.append(ws)
except Exception as e:
with lock:
results.append(e)
threads = [threading.Thread(target=_create) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
# Manager holds at most max_active; overflow creates must raise
# (5 overflows raise, 3 succeed — no silent exceed).
assert mgr.count <= 3
successes = [r for r in results if isinstance(r, Workstream)]
assert len(successes) <= 3
# ---------------------------------------------------------------------------
# open — lazy rehydrate
# ---------------------------------------------------------------------------
def test_open_returns_none_for_missing_row() -> None:
mgr, _, _ = _make_manager()
opened = mgr.open("missing")
assert opened is None
def test_open_blocks_deleted_state() -> None:
mgr, _, storage = _make_manager()
ws = mgr.create(user_id="u1")
mgr.close(ws.id)
# Flip the row to the tombstone state — open must refuse it.
storage.rows[ws.id].state = "deleted"
opened = mgr.open(ws.id)
assert opened is None
def test_open_resurrects_closed_state() -> None:
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
assert mgr.get(ws_id) is None
reopened = mgr.open(ws_id)
assert reopened is not None
assert reopened.id == ws_id
assert reopened.session is not None
assert reopened.session.resumed is True # type: ignore[attr-defined]
# Open fires emit_rehydrated (NOT emit_created) so observers can
# gate any extra resurrect-only setup on it (e.g. coord's
# storage-seeded children rebuild).
assert ws_id in [e.ws_id for e in adapter.events_of("rehydrated")]
def test_open_supports_tokenless_legacy_rows_but_hides_creating() -> None:
"""Rehydrate needs only the public row; private create state stays hidden."""
rows = {
"legacy-closed": {
"ws_id": "legacy-closed",
"user_id": "u1",
"name": "legacy",
"kind": WorkstreamKind.INTERACTIVE,
"state": "closed",
"parent_ws_id": None,
"project_id": None,
"persona": "",
},
"pending-create": {
"ws_id": "pending-create",
"user_id": "u1",
"name": "pending",
"kind": WorkstreamKind.INTERACTIVE,
"state": "creating",
"parent_ws_id": None,
"project_id": None,
"persona": "",
},
}
class _LegacyStorage:
"""Pre-incarnation read surface: deliberately has no private snapshot API."""
def get_workstream(self, ws_id: str) -> dict[str, Any] | None:
return rows.get(ws_id)
def load_workstream_config(self, ws_id: str) -> dict[str, str]:
return {}
def touch_workstream(self, ws_id: str) -> None:
return None
adapter = FakeAdapter()
mgr = SessionManager(
adapter,
storage=_LegacyStorage(), # type: ignore[arg-type]
max_active=2,
event_emitter=adapter,
)
reopened = mgr.open("legacy-closed")
assert reopened is not None
assert reopened._fork_reservation_token == ""
assert mgr.open("pending-create") is None
assert mgr.get("pending-create") is None
def test_open_threads_saved_model_alias_into_build_session() -> None:
"""Reopening a closed ws must build the session with the *original*
model alias, not the current registry default.
Without this, ``build_session(ws)`` is called with ``model=None`` →
the production session_factory resolves ``_effective_default_alias()``
→ ChatSession's ``__init__`` writes those defaults to
``workstream_config`` (INSERT OR REPLACE) → the subsequent
``resume()`` restores what is now the default. Net effect: every
persisted knob (model, temperature, reasoning_effort, max_tokens,
skill, the persona stamp, instructions, …) silently resets on every
reopen and on every service restart.
"""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
# Pretend the user set a non-default alias when the ws was created;
# the real path goes through ChatSession._save_config but the
# FakeSession in this suite doesn't model that, so seed directly.
storage.ws_config[ws_id] = {"model_alias": "gpt-5-pro"}
mgr.close(ws_id)
adapter.last_build_model = "<unset>" # sentinel — must be overwritten
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.last_build_model == "gpt-5-pro"
def test_open_drops_saved_alias_when_validator_rejects() -> None:
"""When the persisted alias is no longer in the registry, the
manager must drop it before reaching ``build_session``. The
factory still raises on unknown aliases on the fresh-create path
(so a typo in body.model surfaces as 503), so the rehydrate path
has to filter the alias here rather than relying on factory-side
fallback. Without this filter, every reopen of a workstream pinned
to a since-removed alias 500s."""
mgr, adapter, storage = _make_manager(
# Validator says "alias is no longer in the registry".
model_validator=lambda alias: False,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": "since-removed-alias"}
mgr.close(ws_id)
adapter.last_build_model = "<unset>"
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.last_build_model is None # alias dropped before reaching build_session
def test_open_keeps_saved_alias_when_validator_accepts() -> None:
"""Sanity: an alias that still resolves must be passed through
unchanged. Filter only fires for stale aliases."""
accepted: list[str] = []
def validator(alias: str) -> bool:
accepted.append(alias)
return True
mgr, adapter, storage = _make_manager(model_validator=validator)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": "still-live"}
mgr.close(ws_id)
adapter.last_build_model = "<unset>"
reopened = mgr.open(ws_id)
assert reopened is not None
# Initial filter plus the pre/post-resume race checks all see the
# same still-live alias.
assert accepted == ["still-live", "still-live", "still-live"]
assert adapter.last_build_model == "still-live"
def test_open_retries_default_when_alias_disappears_during_build() -> None:
"""The validator -> factory straddle retries only the rehydrate build."""
saved_alias = "raced-away"
live_aliases = {saved_alias}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_alias}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.built_sessions.clear()
adapter.cleaned_up.clear()
def build(ws: Workstream, model: object | None) -> FakeSession:
if model == saved_alias:
live_aliases.clear()
raise UnknownModelAliasError(saved_alias)
return FakeSession(ws.id)
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
active: Any = reopened.session
ui: Any = reopened.ui
assert adapter.build_models == [saved_alias, None]
assert active is adapter.built_sessions[-1]
assert active.resumed is True
assert adapter.cleaned_up == []
assert reopened._closed is False
assert ui.closed_broadcast is False
def test_open_does_not_retry_when_alias_recheck_fails() -> None:
"""A validator outage is not proof that the saved alias disappeared."""
saved_alias = "indeterminate"
checks = 0
def validator(alias: str) -> bool:
nonlocal checks
assert alias == saved_alias
checks += 1
if checks == 1:
return True
raise RuntimeError("registry membership unavailable")
mgr, adapter, storage = _make_manager(model_validator=validator)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_alias}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.cleaned_up.clear()
def build(_ws: Workstream, model: object | None) -> FakeSession:
if model == saved_alias:
raise UnknownModelAliasError(saved_alias)
return FakeSession(ws_id)
adapter.build_session_hook = build
with pytest.raises(ValueError, match=f"Unknown model alias: {saved_alias}"):
mgr.open(ws_id)
assert checks == 2
assert adapter.build_models == [saved_alias]
assert adapter.cleaned_up == [ws_id]
assert mgr.get(ws_id) is None
@pytest.mark.parametrize(
"failure",
[
pytest.param(
RuntimeError("unrelated factory failure"),
id="other-failure",
),
pytest.param(
ValueError("unrelated factory value failure"),
id="other-value-failure",
),
pytest.param(
ModelClientConstructionError("client construction failed"),
id="client-construction",
),
],
)
def test_open_does_not_downgrade_non_alias_build_failure(failure: Exception) -> None:
saved_alias = "broken-but-saved"
live_aliases = {saved_alias}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_alias}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.cleaned_up.clear()
def build(_ws: Workstream, model: object | None) -> FakeSession:
if model == saved_alias:
# Even a coincident removal must not erase a more specific
# construction/runtime failure.
live_aliases.clear()
raise failure
return FakeSession(ws_id)
adapter.build_session_hook = build
with pytest.raises(type(failure), match=str(failure)):
mgr.open(ws_id)
assert adapter.build_models == [saved_alias]
assert adapter.cleaned_up == [ws_id]
assert mgr.get(ws_id) is None
def test_open_replaces_candidate_when_alias_disappears_before_resume() -> None:
"""A constructed stale lane is closed without closing its shared UI."""
saved_alias = "gone-after-build"
live_aliases = {saved_alias}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_alias}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.built_sessions.clear()
adapter.cleaned_up.clear()
stale: list[FakeSession] = []
def build(ws: Workstream, model: object | None) -> FakeSession:
session = FakeSession(ws.id)
if model == saved_alias:
stale.append(session)
live_aliases.clear()
return session
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
active: Any = reopened.session
ui: Any = reopened.ui
assert adapter.build_models == [saved_alias, None]
assert len(stale) == 1
assert stale[0].closed is True
assert stale[0].cancelled is True
assert stale[0].resumed is False
assert active is adapter.built_sessions[-1]
assert active.resumed is True
assert active.closed is False
assert adapter.cleaned_up == []
assert reopened._closed is False
assert ui.closed_broadcast is False
def test_open_replaces_candidate_when_alias_disappears_during_resume() -> None:
"""The post-resume check catches the has_alias -> bind race."""
saved_alias = "gone-during-resume"
live_aliases = {saved_alias}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_alias}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.built_sessions.clear()
adapter.cleaned_up.clear()
stale: list[FakeSession] = []
def build(ws: Workstream, model: object | None) -> FakeSession:
session = FakeSession(ws.id)
if model == saved_alias:
stale.append(session)
session.resume_hook = live_aliases.clear
return session
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
active: Any = reopened.session
ui: Any = reopened.ui
assert adapter.build_models == [saved_alias, None]
assert len(stale) == 1
assert stale[0].resumed is True
assert stale[0].closed is True
assert stale[0].cancelled is True
assert active is adapter.built_sessions[-1]
assert active.resumed is True
assert active.closed is False
assert adapter.cleaned_up == []
assert reopened._closed is False
assert ui.closed_broadcast is False
def test_open_validates_alias_that_resume_actually_adopts() -> None:
"""Post-resume validation must not reuse the factory candidate alias."""
saved_before = "saved-before"
saved_during = "saved-during"
live_aliases = {saved_before, saved_during}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
storage.ws_config[ws_id] = {"model_alias": saved_before}
mgr.close(ws_id)
adapter.build_models.clear()
adapter.built_sessions.clear()
stale: list[FakeSession] = []
def build(_ws: Workstream, model: object | None) -> FakeSession:
alias = saved_before if model is None else str(model)
session = FakeSession(ws_id, model_alias=alias)
if model == saved_before:
stale.append(session)
def adopt_then_remove() -> None:
session.model_alias = saved_during
live_aliases.remove(saved_during)
session.resume_hook = adopt_then_remove
return session
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.build_models == [saved_before, None]
assert stale[0].resumed is True
assert stale[0].cancelled is True
assert stale[0].closed is True
assert reopened.session is adapter.built_sessions[-1]
assert reopened.session.model_alias == saved_before # type: ignore[union-attr]
def test_open_falls_back_to_none_when_no_saved_alias() -> None:
"""Reopening a ws with no saved alias must pass ``model=None`` to
``build_session`` so the adapter's session_factory can fall back to
the current default — matching the user's intent: best effort
restore, default when the original is gone."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
# No ws_config row — simulates "alias was never saved" or "saved
# alias was empty string".
assert ws_id not in storage.ws_config
mgr.close(ws_id)
adapter.last_build_model = "<unset>"
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.last_build_model is None
def test_open_retries_when_factory_default_disappears_during_resolution() -> None:
"""A ``model=None`` factory race gets the same exact alias-miss retry."""
live_aliases = {"default-a"}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
adapter.build_models.clear()
def build(_ws: Workstream, model: object | None) -> FakeSession:
assert model is None
if len(adapter.build_models) == 1:
live_aliases.clear()
live_aliases.add("default-b")
raise UnknownModelAliasError("default-a")
return FakeSession(ws_id, model_alias="default-b")
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.build_models == [None, None]
assert reopened.session is adapter.built_sessions[-1]
assert reopened.session.model_alias == "default-b" # type: ignore[union-attr]
def test_open_replaces_default_candidate_removed_before_resume() -> None:
"""The concrete default alias is rechecked even with no persisted alias."""
live_aliases = {"default-a"}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
adapter.build_models.clear()
stale: list[FakeSession] = []
def build(_ws: Workstream, model: object | None) -> FakeSession:
assert model is None
alias = next(iter(live_aliases))
session = FakeSession(ws_id, model_alias=alias)
if alias == "default-a":
stale.append(session)
live_aliases.clear()
live_aliases.add("default-b")
return session
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.build_models == [None, None]
assert stale[0].cancelled is True
assert stale[0].closed is True
assert reopened.session is adapter.built_sessions[-1]
assert reopened.session.model_alias == "default-b" # type: ignore[union-attr]
def test_open_replaces_default_candidate_removed_during_resume() -> None:
"""Post-resume validation follows the candidate alias, not saved None."""
live_aliases = {"default-a"}
mgr, adapter, storage = _make_manager(
model_validator=lambda alias: alias in live_aliases,
)
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
adapter.build_models.clear()
stale: list[FakeSession] = []
def build(_ws: Workstream, model: object | None) -> FakeSession:
assert model is None
alias = next(iter(live_aliases))
session = FakeSession(ws_id, model_alias=alias)
if alias == "default-a":
stale.append(session)
def switch_default() -> None:
live_aliases.clear()
live_aliases.add("default-b")
session.resume_hook = switch_default
return session
adapter.build_session_hook = build
reopened = mgr.open(ws_id)
assert reopened is not None
assert adapter.build_models == [None, None]
assert stale[0].resumed is True
assert stale[0].cancelled is True
assert stale[0].closed is True
assert reopened.session is adapter.built_sessions[-1]
assert reopened.session.model_alias == "default-b" # type: ignore[union-attr]
def test_open_touches_workstream_on_rehydrate() -> None:
"""Rehydrating a workstream must bump its ``updated`` so a concurrent
close_idle pass-2 in this same process can't clobber the freshly-loaded
row to ``closed`` because its DB ``updated`` is older than the cutoff.
The touch is best-effort (try/except in open()) but must fire on the
happy path."""
mgr, _, storage = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
storage.touch_calls.clear() # only care about touches from rehydrate
reopened = mgr.open(ws_id)
assert reopened is not None
assert ws_id in storage.touch_calls
def test_open_ignores_owner_mismatch() -> None:
# Turnstone is a trusted-team tool; row-level ownership is
# metadata, not an access boundary. ``open`` no longer cares
# who the caller is — any authenticated caller can rehydrate
# any persisted workstream. Scope-level auth at the HTTP
# layer is the only gate.
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1")
mgr.close(ws.id)
reopened = mgr.open(ws.id)
assert reopened is not None
assert reopened.user_id == "u1" # metadata preserved
def test_open_rejects_wrong_kind() -> None:
mgr, _, storage = _make_manager()
ws = mgr.create(user_id="u1")
# Storage row claims a different kind than our adapter's.
storage.rows[ws.id].kind = WorkstreamKind.COORDINATOR.value
mgr.close(ws.id)
opened = mgr.open(ws.id)
assert opened is None
def test_concurrent_open_for_same_ws_id_returns_same_session() -> None:
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
mgr.close(ws_id)
adapter.build_session_calls = 0
adapter.build_session_delay = 0.02
results: list[Workstream | None] = []
lock = threading.Lock()
def _open() -> None:
r = mgr.open(ws_id)
with lock:
results.append(r)
threads = [threading.Thread(target=_open) for _ in range(6)]
for t in threads:
t.start()
for t in threads:
t.join()
loaded = [r for r in results if r is not None]
assert len(loaded) == 6
# All threads got the same Workstream instance — no duplicate session.
sessions = {id(r.session) for r in loaded}
assert len(sessions) == 1
# build_session ran exactly once — the per-ws lock serialized the rest.
assert adapter.build_session_calls == 1
# ---------------------------------------------------------------------------
# cancel
# ---------------------------------------------------------------------------
def test_cancel_resolves_all_parallel_approval_cycles() -> None:
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1")
ui = MagicMock()
ws.ui = ui
assert mgr.cancel(ws.id) is True
assert ws.session.cancelled is True # type: ignore[attr-defined]
ui.resolve_all_approvals.assert_called_once_with(False, "cancelled")
ui.resolve_approval.assert_not_called()
def test_cancel_falls_back_to_legacy_single_approval_api() -> None:
class _LegacyApprovalUI:
def __init__(self) -> None:
self.resolutions: list[tuple[bool, str]] = []
def resolve_approval(self, approved: bool, feedback: str) -> None:
self.resolutions.append((approved, feedback))
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1")
ui = _LegacyApprovalUI()
ws.ui = ui
assert mgr.cancel(ws.id) is True
assert ws.session.cancelled is True # type: ignore[attr-defined]
assert ui.resolutions == [(False, "cancelled")]
# ---------------------------------------------------------------------------
# close
# ---------------------------------------------------------------------------
def test_close_pending_fork_deletes_its_reserved_storage_row() -> None:
mgr, _, storage = _make_manager()
ws = mgr.create(
user_id="u1",
defer_emit_created=True,
_fork_reservation=True,
)
assert ws._fork_reservation_token
assert storage.fork_reservations[ws.id] == ws._fork_reservation_token
assert mgr.close(ws.id) is True
assert ws.id not in storage.rows
assert ws.id not in storage.fork_reservations
assert (ws.id, "closed") not in storage.state_updates
def test_close_pending_fork_does_not_delete_replacement_reservation() -> None:
mgr, _, storage = _make_manager()
ws = mgr.create(
user_id="u1",
defer_emit_created=True,
_fork_reservation=True,
)
storage.rows[ws.id].name = "replacement"
storage.fork_reservations[ws.id] = "replacement-incarnation"
assert mgr.close(ws.id) is True
assert storage.rows[ws.id].name == "replacement"
assert storage.fork_reservations[ws.id] == "replacement-incarnation"
assert (ws.id, "closed") not in storage.state_updates
def test_close_unblocks_ui_and_emits_closed() -> None:
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
ws_id = ws.id
closed = mgr.close(ws_id)
assert closed is True
assert mgr.get(ws_id) is None
assert ws_id in adapter.cleaned_up
assert ws_id in [e.ws_id for e in adapter.events_of("closed")]
# Storage reflects the close.
assert (ws_id, "closed") in storage.state_updates
# UI events unblocked.
assert ws.ui.approval_unblocked is True # type: ignore[attr-defined]
assert ws.ui.plan_unblocked is True # type: ignore[attr-defined]
assert ws.ui.closed_broadcast is True # type: ignore[attr-defined]
# Session cancelled + closed.
assert ws.session.cancelled is True # type: ignore[attr-defined]
assert ws.session.closed is True # type: ignore[attr-defined]
def test_close_last_workstream_succeeds() -> None:
"""The WSM 'refuse to close last workstream' guard is gone (#400 follow-up).
The default startup workstream relic is deleted; the dashboard
handles the 0-workstream state and callers can close freely.
"""
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1")
closed = mgr.close(ws.id)
assert closed is True
assert mgr.count == 0
def test_close_unknown_returns_false() -> None:
mgr, _, _ = _make_manager()
closed = mgr.close("not-there")
assert closed is False
# ---------------------------------------------------------------------------
# delete — hard-delete event broadcast (storage delete is the caller's job)
# ---------------------------------------------------------------------------
def test_delete_emits_closed_with_reason_deleted() -> None:
"""Hard-delete a still-loaded workstream: the in-memory slot
drops AND a ``ws_closed`` event fires with ``reason='deleted'``
so the cluster collector → coord adapter chain can re-emit as
``child_ws_closed`` and the operator's child-tree drops the row.
Pre-fix the storage row vanished but no event fired, so a
long-lived dashboard tab would leave the deleted child visible
(with its last-known state) until a full reload."""
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", name="will-be-deleted")
ws_id = ws.id
deleted = mgr.delete(ws_id)
assert deleted is True
# In-memory slot released — capacity restored just like close.
assert mgr.get(ws_id) is None
# Closed event fired with the deletion reason.
closed_events = adapter.events_of("closed")
assert len(closed_events) == 1
ev = closed_events[0]
assert ev.ws_id == ws_id
assert ev.reason == "deleted"
assert ev.name == "will-be-deleted"
# UI cleanup ran (mirrors close ordering).
assert ws_id in adapter.cleaned_up
def test_delete_unloaded_ws_still_fires_event() -> None:
"""A row that was closed (and therefore unloaded from memory)
before delete still needs the broadcast — otherwise a closed
row that's then deleted leaves the closed-state child stuck on
the dashboard tree forever. The in-memory return is False
(nothing to release) but the event MUST fire."""
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", name="closed-then-deleted")
ws_id = ws.id
mgr.close(ws_id)
# Drain the close event so we can assert the delete event in isolation.
pre_delete_closed = list(adapter.events_of("closed"))
assert len(pre_delete_closed) == 1
assert pre_delete_closed[0].reason == "closed"
# Deleting an already-unloaded ws — no in-memory slot to release.
result = mgr.delete(ws_id, name="closed-then-deleted")
assert result is False
# But the event MUST fire — the dashboard hasn't seen the row drop yet.
closed_events = adapter.events_of("closed")
assert len(closed_events) == 2
delete_event = closed_events[1]
assert delete_event.ws_id == ws_id
assert delete_event.reason == "deleted"
assert delete_event.name == "closed-then-deleted"
def test_delete_falls_back_to_workstream_name_when_caller_omits() -> None:
"""When the caller doesn't snapshot a name, the event payload
falls back to the live workstream's name so operator toasts on
the global queue still carry useful context."""
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1", name="auto-name-from-ws")
mgr.delete(ws.id) # no name kwarg
closed_events = adapter.events_of("closed")
assert closed_events[-1].name == "auto-name-from-ws"
def test_delete_returns_false_for_unknown_ws_id() -> None:
"""Idempotent on an absent + never-loaded ws_id — still fires
the event so a dashboard with a stale row can drop it, and
returns False so the caller knows nothing was tracked."""
mgr, adapter, _ = _make_manager()
result = mgr.delete("never-existed")
assert result is False
# Event still fires — a stale dashboard entry is exactly the
# case where this matters.
closed_events = adapter.events_of("closed")
assert len(closed_events) == 1
assert closed_events[0].ws_id == "never-existed"
assert closed_events[0].reason == "deleted"
def test_delete_without_event_emitter_is_quiet() -> None:
"""When the manager is constructed without an event emitter
(e.g. the no-op kind in tests), ``delete`` releases the slot
silently — the no-emitter branch must not raise."""
mgr, adapter, _ = _make_manager(event_emitter=None)
ws = mgr.create(user_id="u1")
result = mgr.delete(ws.id)
assert result is True
# Adapter still gets cleanup_ui — that's the kind-side surface,
# distinct from the lifecycle-event side channel.
assert ws.id in adapter.cleaned_up
# ---------------------------------------------------------------------------
# set_state
# ---------------------------------------------------------------------------
def test_set_state_updates_storage_and_fires_observer() -> None:
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
mgr.set_state(ws.id, WorkstreamState.RUNNING)
assert ws.state == WorkstreamState.RUNNING
assert (ws.id, WorkstreamState.RUNNING.value) in storage.state_updates
state_events = adapter.events_of("state")
assert any(e.ws_id == ws.id and e.state == WorkstreamState.RUNNING for e in state_events)
def test_set_state_unknown_ws_is_noop() -> None:
mgr, adapter, _ = _make_manager()
mgr.set_state("ghost", WorkstreamState.RUNNING)
assert adapter.events_of("state") == []
def test_set_state_deferred_mutates_live_then_persists_before_publish() -> None:
"""The split path mutates now and defers its ordered durable/observer tail."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
order: list[tuple[str, str]] = []
deferred: list[Any] = []
before = ws.last_active
storage.update_workstream_state = MagicMock(
side_effect=lambda _ws_id, state: order.append(("persist", state))
)
adapter.emit_state = MagicMock(
side_effect=lambda _ws, state: order.append(("adapter", state.value))
)
mgr.subscribe_to_state(lambda _ws_id, state: order.append(("subscriber", state.value)))
mgr.set_state_deferred(
ws.id,
WorkstreamState.RUNNING,
error_msg="live detail",
deferred_persistence=deferred,
)
assert ws.state is WorkstreamState.RUNNING
assert ws.error_message == "live detail"
assert ws.last_active >= before
assert order == []
assert len(deferred) == 1
deferred[0]()
assert order == [
("persist", "running"),
("adapter", "running"),
("subscriber", "running"),
]
def test_direct_set_state_persists_before_publishing() -> None:
"""Legacy/direct callers retain the durable-before-live ordering."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
order: list[tuple[str, str]] = []
storage.update_workstream_state = MagicMock(
side_effect=lambda _ws_id, state: order.append(("persist", state))
)
adapter.emit_state = MagicMock(
side_effect=lambda _ws, state: order.append(("publish", state.value))
)
mgr.set_state(ws.id, WorkstreamState.ERROR, error_msg="boom")
assert ws.state is WorkstreamState.ERROR
assert ws.error_message == "boom"
assert order == [("persist", "error"), ("publish", "error")]
def test_direct_successor_waits_for_running_deferred_tail_and_publishes_last() -> None:
"""Direct and deferred callers share one persistence/publication lane."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
adapter.events.clear()
old_write_started = threading.Event()
release_old_write = threading.Event()
writes: list[str] = []
def update_state(_ws_id: str, state: str) -> None:
if state == "running":
old_write_started.set()
assert release_old_write.wait(2)
writes.append(state)
storage.update_workstream_state = update_state # type: ignore[method-assign]
deferred: list[Callable[[], None]] = []
assert mgr.set_state_deferred(
ws.id,
WorkstreamState.RUNNING,
deferred_persistence=deferred,
)
predecessor = threading.Thread(target=deferred[0])
successor = threading.Thread(
target=mgr.set_state,
args=(ws.id, WorkstreamState.IDLE),
)
predecessor.start()
try:
assert old_write_started.wait(2)
successor.start()
deadline = time.monotonic() + 1
while ws.state is not WorkstreamState.IDLE and time.monotonic() < deadline:
time.sleep(0.005)
assert ws.state is WorkstreamState.IDLE
assert successor.is_alive()
assert adapter.events_of("state") == []
finally:
release_old_write.set()
predecessor.join(2)
if successor.ident is not None:
successor.join(2)
assert not predecessor.is_alive()
assert not successor.is_alive()
assert writes == ["running", "idle"]
assert [event.state for event in adapter.events_of("state")] == [WorkstreamState.IDLE]
def test_deferred_state_without_emitter_still_publishes_to_subscriber() -> None:
"""No emitter is a valid accepted tail, not the stale-tail sentinel."""
mgr, _, _ = _make_manager(event_emitter=None)
ws = mgr.create(user_id="u1")
observed: list[WorkstreamState] = []
mgr.subscribe_to_state(lambda _ws_id, state: observed.append(state))
deferred: list[Callable[[], None]] = []
assert mgr.set_state_deferred(
ws.id,
WorkstreamState.RUNNING,
deferred_persistence=deferred,
)
deferred[0]()
assert observed == [WorkstreamState.RUNNING]
def test_delayed_state_persistence_cannot_overwrite_closed_workstream() -> None:
"""A close tombstone makes an already-returned state closure wholly inert."""
mgr, adapter, storage = _make_manager()
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
deferred: list[Any] = []
subscriber_events: list[tuple[str, WorkstreamState]] = []
mgr.subscribe_to_state(lambda ws_id, state: subscriber_events.append((ws_id, state)))
mgr.set_state_deferred(
ws.id,
WorkstreamState.RUNNING,
deferred_persistence=deferred,
)
assert ws.state is WorkstreamState.RUNNING
assert adapter.events_of("state") == []
assert storage.state_updates == []
assert len(deferred) == 1
assert mgr.close(ws.id) is True
assert ws._closed is True
assert storage.state_updates == [(ws.id, "closed")]
deferred[0]()
assert storage.state_updates == [(ws.id, "closed")]
assert storage.rows[ws.id].state == "closed"
assert adapter.events_of("state") == []
assert subscriber_events == []
# ---------------------------------------------------------------------------
# close_idle / list_all / get / count
# ---------------------------------------------------------------------------
def test_reap_stale_creating_reservations_recovers_local_restart_and_dead_peer() -> None:
mgr, _, storage = _make_manager(node_id="stable-node")
storage.live_services["server"] = ["stable-node", "live-peer"]
for ws_id, node_id, token in [
("abandoned-local", "stable-node", "local-token"),
("abandoned-dead-peer", "dead-peer", "dead-token"),
("protected-live-peer", "live-peer", "live-token"),
("ambiguous-tokenless", "dead-peer", ""),
]:
storage.register_workstream(
ws_id,
node_id=node_id,
kind=WorkstreamKind.INTERACTIVE,
state="creating",
updated="2020-01-01T00:00:00",
fork_reservation_token=token,
)
storage.register_workstream(
"already-published",
node_id="dead-peer",
kind=WorkstreamKind.INTERACTIVE,
state="idle",
updated="2020-01-01T00:00:00",
fork_reservation_token="published-token",
)
pending = mgr.create(user_id="u1", defer_emit_created=True)
storage.rows[pending.id].updated = "2020-01-01T00:00:00"
reaped = mgr.reap_stale_creating_reservations(max_age_seconds=0)
assert set(reaped) == {
"abandoned-local",
"abandoned-dead-peer",
"ambiguous-tokenless",
}
assert "protected-live-peer" in storage.rows
assert "ambiguous-tokenless" not in storage.rows
assert storage.rows["already-published"].state == "idle"
assert storage.rows[pending.id].state == "creating"
assert mgr.commit_create(pending) is True
def test_reap_stale_creating_reservations_fails_closed_on_liveness_error() -> None:
mgr, _, storage = _make_manager(node_id="stable-node")
storage.list_services_raises = True
storage.register_workstream(
"ambiguous-owner",
node_id="stable-node",
kind=WorkstreamKind.INTERACTIVE,
state="creating",
updated="2020-01-01T00:00:00",
fork_reservation_token="reservation",
)
assert mgr.reap_stale_creating_reservations(max_age_seconds=0) == []
assert storage.rows["ambiguous-owner"].state == "creating"
def test_reap_stale_creating_reservations_fails_closed_on_delete_error() -> None:
mgr, _, storage = _make_manager(node_id="stable-node")
storage.delete_stale_creating_raises = True
storage.register_workstream(
"storage-uncertain",
node_id="stable-node",
kind=WorkstreamKind.INTERACTIVE,
state="creating",
updated="2020-01-01T00:00:00",
fork_reservation_token="reservation",
)
assert mgr.reap_stale_creating_reservations(max_age_seconds=0) == []
assert storage.rows["storage-uncertain"].state == "creating"
def test_reap_stale_creating_reservations_supports_node_less_cli_boot() -> None:
mgr, _, storage = _make_manager(node_id=None)
storage.live_services["server"] = ["live-server"]
storage.register_workstream(
"abandoned-cli-create",
node_id=None,
kind=WorkstreamKind.INTERACTIVE,
state="creating",
updated="2020-01-01T00:00:00",
fork_reservation_token="cli-reservation",
)
storage.register_workstream(
"remote-live-create",
node_id="live-server",
kind=WorkstreamKind.INTERACTIVE,
state="creating",
updated="2020-01-01T00:00:00",
fork_reservation_token="remote-reservation",
)
assert mgr.reap_stale_creating_reservations(max_age_seconds=0) == ["abandoned-cli-create"]
assert "remote-live-create" in storage.rows
def test_close_idle_closes_old_idle_and_keeps_active() -> None:
mgr, _, _ = _make_manager()
old = mgr.create(user_id="u1")
fresh = mgr.create(user_id="u1")
running = mgr.create(user_id="u1")
old.last_active = time.monotonic() - 100
running.state = WorkstreamState.RUNNING
running.last_active = time.monotonic() - 100
closed = mgr.close_idle(max_age_seconds=10.0)
assert old.id in closed
assert mgr.get(old.id) is None
# Fresh stays (not old enough).
assert mgr.get(fresh.id) is not None
# Running stays (wrong state).
assert mgr.get(running.id) is not None
def test_close_idle_on_empty_manager_returns_empty_list() -> None:
mgr, _, _ = _make_manager()
assert mgr.close_idle(max_age_seconds=1.0) == []
def test_close_idle_runs_db_orphan_pass() -> None:
"""DB rows of this kind that aren't loaded into the manager get
bulk-closed when their ``updated`` is older than the cutoff. Catches
the orphan-after-process-restart case the original close_idle missed."""
mgr, _, storage = _make_manager()
# Orphan rows live in storage but were never loaded via mgr.create.
storage.register_workstream(
"orphan-1",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
storage.register_workstream(
"orphan-2",
kind=WorkstreamKind.INTERACTIVE,
state="thinking",
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert set(closed) == {"orphan-1", "orphan-2"}
assert ("orphan-1", "closed") in storage.state_updates
assert ("orphan-2", "closed") in storage.state_updates
assert storage.rows["orphan-1"].state == "closed"
assert storage.rows["orphan-2"].state == "closed"
def test_close_idle_excludes_loaded_workstreams_from_db_pass() -> None:
"""A workstream loaded into memory must NOT be reaped by the DB
orphan pass even when its storage ``updated`` is stale — the
in-memory pass owns those. Verifies the exclude_ws_ids plumbing."""
mgr, _, storage = _make_manager()
ws = mgr.create(user_id="u1")
# Force the storage row's ``updated`` to look stale. In practice
# ``set_state`` would bump it, but we're simulating a long-running
# active workstream whose updated drifted older than the cutoff.
storage.rows[ws.id].updated = "2020-01-01T00:00:00"
# Huge timeout so the in-memory IDLE pass skips it (stays loaded).
closed = mgr.close_idle(max_age_seconds=10_000.0)
assert ws.id not in closed
assert mgr.get(ws.id) is not None
assert storage.rows[ws.id].state == "idle"
def test_close_idle_filters_db_orphans_by_kind() -> None:
"""An interactive manager's close_idle must not touch coordinator
rows in storage and vice versa. Without this filter, both managers
would race to close each other's rows."""
mgr, _, storage = _make_manager() # interactive by default
storage.register_workstream(
"coord-orphan",
kind=WorkstreamKind.COORDINATOR,
updated="2020-01-01T00:00:00",
)
storage.register_workstream(
"interactive-orphan",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert "interactive-orphan" in closed
assert "coord-orphan" not in closed
assert storage.rows["coord-orphan"].state == "idle"
assert storage.rows["interactive-orphan"].state == "closed"
def test_close_idle_protects_rows_owned_by_live_services() -> None:
"""Multi-node correctness: rows whose ``node_id`` matches a service
with a recent heartbeat must NOT be reaped, even when *this* manager
is on a different node — the alive peer may legitimately have them
loaded. Liveness is the rendezvous router's primitive (post-PR-#384);
using it here keeps reap scoping aligned with routing.
Default ``_make_manager`` uses an INTERACTIVE adapter, which derives
``service_type='server'`` — so live_services seeded under "server"
are what the manager queries."""
mgr, _, storage = _make_manager()
storage.live_services["server"] = ["node-b"] # only node-b is alive
storage.register_workstream(
"ours-from-dead-node",
node_id="node-a", # dead pod (not in live_services)
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
storage.register_workstream(
"theirs-still-alive",
node_id="node-b",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert closed == ["ours-from-dead-node"]
assert storage.rows["ours-from-dead-node"].state == "closed"
assert storage.rows["theirs-still-alive"].state == "idle"
def test_close_idle_protects_live_services_for_coordinator_kind() -> None:
"""Coord-side parity: a coordinator manager derives
``service_type='console'``, so live_services seeded under "console"
are what gets queried. Mirrors the interactive test to ensure both
halves of the production wiring are exercised."""
coord_adapter = FakeAdapter(kind=WorkstreamKind.COORDINATOR)
mgr, _, storage = _make_manager(coord_adapter)
storage.live_services["console"] = ["console"] # console is alive
storage.register_workstream(
"alive-console-coord",
node_id="console",
kind=WorkstreamKind.COORDINATOR,
updated="2020-01-01T00:00:00",
)
storage.register_workstream(
"dead-console-coord",
node_id="dead-console-instance", # not in live set
kind=WorkstreamKind.COORDINATOR,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert closed == ["dead-console-coord"]
assert storage.rows["alive-console-coord"].state == "idle"
assert storage.rows["dead-console-coord"].state == "closed"
def test_close_idle_reaps_rows_with_null_node_id() -> None:
"""A row with no ``node_id`` has no owner identity — age alone gates
the reap. Defends against a NULL silently propagating through ``NOT
IN (live)`` and protecting orphans forever."""
mgr, _, storage = _make_manager()
storage.live_services["server"] = ["node-a"]
storage.register_workstream(
"no-owner",
node_id=None,
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert closed == ["no-owner"]
def test_close_idle_reaps_all_orphans_when_no_peers_alive() -> None:
"""When ``list_services`` returns an empty list (no heartbeating
peers), every stale orphan is unprotected and gets reaped. This is
the cold-start / single-process / dead-cluster-recovery case."""
mgr, _, storage = _make_manager()
# storage.live_services["server"] left empty — no peers heartbeating
storage.register_workstream(
"any-node-1",
node_id="node-a",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
storage.register_workstream(
"any-node-2",
node_id="node-b",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert set(closed) == {"any-node-1", "any-node-2"}
def test_close_idle_skips_pass_2_when_list_services_fails() -> None:
"""Conservative fallback: if list_services fails we can't enumerate
live owners safely, so pass 2 must skip rather than reap blind. Pass
1 (in-memory IDLE) still runs."""
mgr, _, storage = _make_manager()
storage.list_services_raises = True
storage.register_workstream(
"would-be-orphan",
node_id="node-a",
kind=WorkstreamKind.INTERACTIVE,
updated="2020-01-01T00:00:00",
)
closed = mgr.close_idle(max_age_seconds=0.0)
assert closed == []
assert storage.rows["would-be-orphan"].state == "idle"
def test_list_all_returns_creation_order() -> None:
mgr, _, _ = _make_manager()
a = mgr.create(user_id="u1")
b = mgr.create(user_id="u1")
c = mgr.create(user_id="u1")
assert [ws.id for ws in mgr.list_all()] == [a.id, b.id, c.id]
def test_count_reflects_live_workstreams() -> None:
mgr, _, _ = _make_manager()
assert mgr.count == 0
a = mgr.create(user_id="u1")
assert mgr.count == 1
mgr.create(user_id="u1")
assert mgr.count == 2
mgr.close(a.id)
assert mgr.count == 1
# ---------------------------------------------------------------------------
# Eviction transport — adapter contract
# ---------------------------------------------------------------------------
def test_eviction_fires_emit_closed_to_adapter_transport() -> None:
mgr, adapter, _ = _make_manager(max_active=2)
a = mgr.create(user_id="u1")
a.last_active = time.monotonic() - 100
mgr.create(user_id="u1")
mgr.create(user_id="u1") # triggers eviction of 'a'
closed_events = adapter.events_of("closed")
evicted = [e for e in closed_events if e.ws_id == a.id]
assert evicted and evicted[0].reason == "evicted"
assert a.id in adapter.cleaned_up
def test_manual_close_uses_closed_reason() -> None:
mgr, adapter, _ = _make_manager()
ws = mgr.create(user_id="u1")
mgr.close(ws.id)
closed_events = [e for e in adapter.events_of("closed") if e.ws_id == ws.id]
assert closed_events and closed_events[0].reason == "closed"
# ---------------------------------------------------------------------------
# CLI focus state
# ---------------------------------------------------------------------------
def test_active_id_seeded_on_first_create() -> None:
mgr, _, _ = _make_manager()
assert mgr.active_id is None
ws = mgr.create(user_id="u1")
assert mgr.active_id == ws.id
assert mgr.get_active() is ws
def test_active_id_unchanged_on_subsequent_creates() -> None:
mgr, _, _ = _make_manager()
first = mgr.create(user_id="u1")
mgr.create(user_id="u1")
# Creating a second workstream doesn't change focus.
assert mgr.active_id == first.id
def test_switch_moves_active_id() -> None:
mgr, _, _ = _make_manager()
a = mgr.create(user_id="u1")
b = mgr.create(user_id="u1")
assert mgr.active_id == a.id
result = mgr.switch(b.id)
assert result is b
assert mgr.active_id == b.id
# Unknown id → no change.
assert mgr.switch("ghost") is None
assert mgr.active_id == b.id
def test_switch_by_index_uses_1_based_ordering() -> None:
mgr, _, _ = _make_manager()
a = mgr.create(user_id="u1")
b = mgr.create(user_id="u1")
c = mgr.create(user_id="u1")
assert mgr.switch_by_index(2) is b
assert mgr.active_id == b.id
assert mgr.switch_by_index(3) is c
assert mgr.switch_by_index(0) is None
assert mgr.switch_by_index(99) is None
# Still on c after the invalid switches.
assert mgr.active_id == c.id
assert mgr.index_of(a.id) == 1
assert mgr.index_of("ghost") == 0
def test_active_id_moves_on_eviction() -> None:
mgr, _, _ = _make_manager(max_active=2)
a = mgr.create(user_id="u1")
a.last_active = time.monotonic() - 100
mgr.create(user_id="u1")
# First create seeded active to a; eviction of a must re-home active.
assert mgr.active_id == a.id
c = mgr.create(user_id="u1") # evicts a
assert mgr.active_id != a.id
assert mgr.active_id in (mgr._order[0], c.id) # type: ignore[attr-defined]
def test_active_id_moves_on_close() -> None:
mgr, _, _ = _make_manager()
a = mgr.create(user_id="u1")
b = mgr.create(user_id="u1")
mgr.switch(a.id)
mgr.close(a.id)
assert mgr.active_id == b.id
def test_eviction_count_tracks_evictions() -> None:
mgr, _, _ = _make_manager(max_active=2)
assert mgr.eviction_count == 0
a = mgr.create(user_id="u1")
a.last_active = time.monotonic() - 100
mgr.create(user_id="u1")
mgr.create(user_id="u1") # evicts a
assert mgr.eviction_count == 1
# ---------------------------------------------------------------------------
# Storage access patterns with mocks (defensive coverage)
# ---------------------------------------------------------------------------
def test_create_uses_configured_node_id() -> None:
storage = MagicMock()
adapter = FakeAdapter()
mgr = SessionManager(
adapter, storage=storage, max_active=3, node_id="node-xyz", event_emitter=adapter
)
mgr.create(user_id="u1")
assert storage.register_workstream.call_args.kwargs["node_id"] == "node-xyz"
# ---------------------------------------------------------------------------
# StateWriter integration — bug-3 invariant under write-behind
# ---------------------------------------------------------------------------
class TestSessionManagerWithStateWriter:
"""When a buffered StateWriter is wired in, ``set_state`` no longer
blocks ``ws._lock`` on a sync DB write — but ``close`` must still
leave 'closed' as the durable final state for the row, never
a buffered transient that flushes after close."""
def _make_with_writer(
self, *, flush_interval: float = 0.05
) -> tuple[SessionManager, FakeStorage, Any]:
from turnstone.core.state_writer import StateWriter
storage = FakeStorage()
writer = StateWriter(storage, flush_interval=flush_interval)
adapter = FakeAdapter()
mgr = SessionManager(
adapter,
storage=storage,
max_active=3,
state_writer=writer,
event_emitter=adapter,
)
return mgr, storage, writer
def test_set_state_buffers_through_writer(self) -> None:
mgr, storage, writer = self._make_with_writer(flush_interval=60.0)
ws = mgr.create(user_id="u1")
# Initial register write may have landed; clear and inspect afterwards.
storage.state_updates.clear()
mgr.set_state(ws.id, WorkstreamState.RUNNING)
# Long flush_interval → buffered, no sync write yet.
assert storage.state_updates == []
# Drain the buffer manually (test stand-in for the periodic flush).
writer.flush()
assert (ws.id, "running") in storage.state_updates
def test_set_state_error_flushes_sync(self) -> None:
"""Terminal ERROR transitions must be durable on return — error
surfacing paths (audit, dashboard) need the row to reflect
the failure before any observer sees it."""
mgr, storage, _writer = self._make_with_writer(flush_interval=60.0)
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
mgr.set_state(ws.id, WorkstreamState.ERROR, error_msg="boom")
# No buffer-drain needed — error path bypasses.
assert (ws.id, "error") in storage.state_updates
def test_close_after_buffered_set_state_writes_closed_not_transient(self) -> None:
"""The bug-3 invariant under write-behind. close() must call
state_writer.discard BEFORE its sync 'closed' write, so any
buffered 'running' state can't be flushed AFTER 'closed'.
"""
mgr, storage, writer = self._make_with_writer(flush_interval=60.0)
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
# Buffer a transient transition.
mgr.set_state(ws.id, WorkstreamState.RUNNING)
# Now close — must drain/discard the buffer + write 'closed' sync.
ok = mgr.close(ws.id)
assert ok is True
# Force a flush; the buffered 'running' must already be gone.
writer.flush()
# The final write for ws.id must be 'closed', and 'running' must
# NOT have landed in storage at all.
ws_writes = [s for w, s in storage.state_updates if w == ws.id]
assert "running" not in ws_writes, f"buffered running flushed after close: {ws_writes}"
assert ws_writes[-1] == "closed"
def test_close_idle_after_buffered_set_state_writes_closed(self) -> None:
"""Same invariant via close_idle (the idle-cleanup batch path)."""
mgr, storage, writer = self._make_with_writer(flush_interval=60.0)
ws = mgr.create(user_id="u1")
storage.state_updates.clear()
mgr.set_state(ws.id, WorkstreamState.RUNNING)
# Force the workstream into IDLE state before close_idle considers
# it (close_idle gates on ws.state, not the buffered state).
with mgr._lock:
mgr._workstreams[ws.id].state = WorkstreamState.IDLE
mgr._workstreams[ws.id].last_active = 0.0 # ancient → stale
closed = mgr.close_idle(max_age_seconds=0)
assert ws.id in closed
writer.flush()
ws_writes = [s for w, s in storage.state_updates if w == ws.id]
assert "running" not in ws_writes, f"buffered running flushed after close_idle: {ws_writes}"
assert ws_writes[-1] == "closed"
def test_set_state_after_close_short_circuits(self) -> None:
"""The existing tombstone (ws._closed) check must still fire
before reaching state_writer.record — set_state after close
must NOT enqueue 'running' to the buffer (which would then
get flushed and resurrect the closed row)."""
mgr, storage, writer = self._make_with_writer(flush_interval=60.0)
ws = mgr.create(user_id="u1")
ws_id = ws.id
# Close first — sets ws._closed=True synchronously.
mgr.close(ws_id)
storage.state_updates.clear()
# A late set_state call (e.g. from a worker still cleaning up).
# Should NOT buffer 'running' for this ws.
mgr.set_state(ws_id, WorkstreamState.RUNNING)
writer.flush()
ws_writes = [s for w, s in storage.state_updates if w == ws_id]
assert "running" not in ws_writes, (
f"set_state after close enqueued through buffer: {ws_writes}"
)
# ---------------------------------------------------------------------------
# Multi-subscriber observer — subscribe_to_state / unsubscribe_from_state
# ---------------------------------------------------------------------------
class TestStateSubscribers:
"""Multi-subscriber observer for ``set_state``.
Used by the CLI's background-attention notifier and by
``SameNodeChildSource``. Subscribe / unsubscribe must be safe under
concurrent dispatch, and dispatch must not skip / repeat callbacks
when subscribers register or unregister mid-iteration.
"""
def test_subscribe_fires_on_set_state(self) -> None:
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1", name="ws", skill=None)
events: list[tuple[str, str]] = []
def cb(ws_id: str, state: WorkstreamState) -> None:
events.append((ws_id, state.value))
mgr.subscribe_to_state(cb)
mgr.set_state(ws.id, WorkstreamState.RUNNING)
assert events == [(ws.id, "running")]
def test_unsubscribe_stops_firing(self) -> None:
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1", name="ws", skill=None)
events: list[str] = []
def cb(_ws_id: str, state: WorkstreamState) -> None:
events.append(state.value)
mgr.subscribe_to_state(cb)
mgr.unsubscribe_from_state(cb)
mgr.set_state(ws.id, WorkstreamState.RUNNING)
assert events == []
def test_unsubscribe_unknown_is_noop(self) -> None:
mgr, _, _ = _make_manager()
# Doesn't raise.
mgr.unsubscribe_from_state(lambda *_: None)
def test_multiple_subscribers_fire_in_registration_order(self) -> None:
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1", name="ws", skill=None)
order: list[int] = []
def make(i: int) -> Callable[[str, WorkstreamState], None]:
def cb(_ws_id: str, _state: WorkstreamState) -> None:
order.append(i)
return cb
mgr.subscribe_to_state(make(1))
mgr.subscribe_to_state(make(2))
mgr.subscribe_to_state(make(3))
mgr.set_state(ws.id, WorkstreamState.IDLE)
assert order == [1, 2, 3]
def test_subscriber_exception_does_not_block_others(self) -> None:
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1", name="ws", skill=None)
survived: list[str] = []
def boom(*_: Any) -> None:
raise RuntimeError("subscriber crash")
def good(_ws_id: str, state: WorkstreamState) -> None:
survived.append(state.value)
mgr.subscribe_to_state(boom)
mgr.subscribe_to_state(good)
mgr.set_state(ws.id, WorkstreamState.RUNNING)
assert survived == ["running"]
@pytest.mark.parametrize("n_threads", [10, 50])
def test_concurrent_subscribe(self, n_threads: int) -> None:
"""Subscribe from many threads; all callbacks land in the list.
Validates the lock around mutation — without it the underlying
list.append could lose entries under contention.
"""
mgr, _, _ = _make_manager()
callbacks = [lambda *_, i=i: None for i in range(n_threads)]
threads = [threading.Thread(target=mgr.subscribe_to_state, args=(cb,)) for cb in callbacks]
for t in threads:
t.start()
for t in threads:
t.join()
# Snapshot under the lock to read the count safely.
with mgr._state_subscribers_lock:
assert len(mgr._state_subscribers) == n_threads
def test_subscribe_during_dispatch_does_not_corrupt_iteration(self) -> None:
"""A subscriber that calls subscribe_to_state during its own
callback must not affect the in-flight dispatch (snapshot
isolation). This is the bug-1 invariant: mutation during
iteration can't shift the iterator's index because dispatch
iterates a snapshot, not the live list.
"""
mgr, _, _ = _make_manager()
ws = mgr.create(user_id="u1", name="ws", skill=None)
fired: list[str] = []
def late(_ws_id: str, state: WorkstreamState) -> None:
fired.append("late:" + state.value)
def first(_ws_id: str, state: WorkstreamState) -> None:
fired.append("first:" + state.value)
mgr.subscribe_to_state(late) # mid-dispatch addition
mgr.subscribe_to_state(first)
mgr.set_state(ws.id, WorkstreamState.RUNNING)
# ``late`` was added during dispatch but the snapshot was
# already frozen — so it doesn't fire on this round.
assert fired == ["first:running"]
# Next round it does fire, in registration order after first.
fired.clear()
mgr.set_state(ws.id, WorkstreamState.IDLE)
assert fired == ["first:idle", "late:idle"]
# ---------------------------------------------------------------------------
# concrete_method — the shared optional-hook probe
# ---------------------------------------------------------------------------
class TestConcreteMethod:
"""Both halves of the shared guard every optional-hook caller drives.
Manager persistence hooks, ``workstream_persistence_state``, the route
layer's cancel / history-handoff / cross-user probes, the nudge watcher's
interjection claim and the coordinator adapter's spawn gate all resolve
their hook through this one helper, so the two semantics below are pinned
once here rather than eleven times at the call sites.
"""
def test_type_defined_method_is_concrete(self) -> None:
class Real:
def hook(self) -> str:
return "real"
found = concrete_method(Real(), "hook")
assert found is not None
assert found() == "real"
def test_missing_method_is_none(self) -> None:
assert concrete_method(object(), "hook") is None
def test_magicmock_auto_vivification_is_not_a_hook(self) -> None:
"""An unconfigured mock answers every attribute with a callable child.
Treating that as a production hook is what the guard exists to stop:
it would make every optional seam look implemented under unit tests.
"""
assert concrete_method(MagicMock(), "hook") is None
def test_instance_dict_hook_is_concrete(self) -> None:
"""A deliberately installed per-instance hook still counts.
This is the half ``session_manager`` had and the route layer's
type-only copies had lost: an explicitly assigned attribute lands in
the instance ``__dict__``, unlike an auto-vivified mock child, so it
is distinguishable and must be honored.
"""
target = MagicMock()
target.hook = lambda: "installed"
found = concrete_method(target, "hook")
assert found is not None
assert found() == "installed"
def test_non_callable_attribute_is_not_a_hook(self) -> None:
class Shadowed:
hook = "not callable"
assert concrete_method(Shadowed(), "hook") is None
def test_slots_object_without_dict_is_supported(self) -> None:
class Slotted:
__slots__ = ()
assert concrete_method(Slotted(), "hook") is None
Reference in New Issue View Git Blame Copy Permalink