From 30746adaa057466123bb724bffee46467ed82edd Mon Sep 17 00:00:00 2001 From: promptclickrun Date: Fri, 18 Sep 2026 02:21:06 +0000 Subject: [PATCH 1/2] Thread mode: coordinator/worker threads (plugin side) Implements the plugin half of the thread-mode protocol contract (PROTOCOL.md): - Four loopdy tools: thread_spawn, thread_status, thread_collect, thread_note - Plugin state under threadmode: and threadmode:worker: keys - Additive pre_llm_call injection (coordinator + worker) and subagent_start/subagent_stop observers - App out-of-turn REST: POST /native/threads/flag, POST /native/threads/register, GET /native/threads/roster - 51 new tests in tests/test_thread_mode.py; full suite matches main baseline --- dashboard/plugin_api.py | 2 + loopdy_plugin/native_threads.py | 157 +++++++ loopdy_plugin/registration.py | 5 + loopdy_plugin/thread_mode.py | 708 ++++++++++++++++++++++++++++++ plugin.yaml | 4 + tests/test_thread_mode.py | 724 +++++++++++++++++++++++++++++++ tests/test_tools_registration.py | 7 + 7 files changed, 1607 insertions(+) create mode 100644 loopdy_plugin/native_threads.py create mode 100644 loopdy_plugin/thread_mode.py create mode 100644 tests/test_thread_mode.py diff --git a/dashboard/plugin_api.py b/dashboard/plugin_api.py index 0e0a446..5d48269 100644 --- a/dashboard/plugin_api.py +++ b/dashboard/plugin_api.py @@ -52,6 +52,7 @@ from loopdy_plugin.room_activity_api import router as room_activity_router # noqa: E402 from loopdy_plugin.native_wiki_api import router as native_wiki_router # noqa: E402 from loopdy_plugin.native_project_git import router as native_project_git_router # noqa: E402 +from loopdy_plugin.native_threads import router as native_threads_router # noqa: E402 from loopdy_plugin.agent_templates import ( # noqa: E402 capability as agent_templates_capability, router as agent_templates_router, @@ -64,6 +65,7 @@ router.include_router(room_activity_router, prefix="/native") router.include_router(native_wiki_router) router.include_router(native_project_git_router) +router.include_router(native_threads_router) router.include_router(agent_templates_router) from loopdy_plugin.managed_notifications_api import router as managed_notifications_router router.include_router(managed_notifications_router) diff --git a/loopdy_plugin/native_threads.py b/loopdy_plugin/native_threads.py new file mode 100644 index 0000000..4a837b5 --- /dev/null +++ b/loopdy_plugin/native_threads.py @@ -0,0 +1,157 @@ +"""App -> plugin REST channel for thread mode (out-of-turn). + +The iOS app manages thread-mode conversations without a live coordinator turn: +it flags a session as thread-mode, registers session-backed worker threads it +created itself (``session.create`` + ``prompt.submit``), and reads the roster. +All state lives in the same PluginState the in-turn tools use, so the roster +stays consistent across both channels. +""" +from __future__ import annotations + +import logging +import sys +import time +from typing import Any, Callable + +from fastapi import APIRouter, Request +from fastapi.responses import Response +from pydantic import BaseModel, ConfigDict, Field + +from .native_api import _NativeRoute, _body, _precondition, _response +from .native_context import NativeAPIError, native_context +from . import thread_mode as _thread_mode + + +logger = logging.getLogger("hermes.plugins.loopdy.thread_mode.rest") +router = APIRouter(prefix="/native/threads", route_class=_NativeRoute) + +# State resolver wiring. The dashboard mounts the bare ``loopdy_plugin`` +# package while Hermes loads the plugin under a profile-safe namespace; the +# plugin registration records its PluginState resolver in the loader-owned +# module and this module scans sys.modules for the matching generation, the +# same pattern native_device_tools uses for its hub. +_registered_profile: str | None = None +_registered_state: Any = None + + +def register_state_resolver(profile: str, state: Any) -> None: + """Called from plugin registration: expose the profile's PluginState.""" + global _registered_profile, _registered_state + _registered_profile = profile + _registered_state = state + + +def _state_for_profile(profile: str) -> Any: + if not isinstance(profile, str) or not profile: + raise NativeAPIError(503, "threads_unavailable", "Thread mode is unavailable.") + for name, module in tuple(sys.modules.items()): + if not name.endswith(".native_threads"): + continue + resolver_profile = getattr(module, "_registered_profile", None) + state = getattr(module, "_registered_state", None) + if resolver_profile == profile and state is not None: + return state + raise NativeAPIError(503, "threads_unavailable", "Thread mode is unavailable.") + + +def _clean_id(value: Any, label: str, maximum: int = 256) -> str: + if not isinstance(value, str) or not value or value != value.strip(): + raise NativeAPIError(422, "invalid_request", f"The {label} is invalid.") + if len(value) > maximum or any(ord(c) < 32 or 127 <= ord(c) <= 159 for c in value): + raise NativeAPIError(422, "invalid_request", f"The {label} is invalid.") + return value + + +class _FlagBody(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + session_id: str = Field(min_length=1, max_length=256) + enabled: bool + + +class _RegisterBody(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + coordinator_session_id: str = Field(min_length=1, max_length=256) + name: str = Field(pattern=r"^[a-z0-9][a-z0-9_-]{0,63}$") + worker_session_id: str = Field(min_length=1, max_length=256) + brief: str = Field(min_length=1, max_length=4000) + + +def _profile_state(request: Request, owner: Any) -> Any: + profile = str(getattr(owner, "serving_profile_id", "") or "") + if not profile: + raise NativeAPIError(503, "threads_unavailable", "Thread mode is unavailable.") + state = _state_for_profile(profile) + if native_context(request) != owner: + raise NativeAPIError(412, "context_changed", "The native context changed; refresh before retrying.") + return state + + +@router.post("/flag") +async def flag(request: Request) -> Response: + owner = native_context(request) + request_id = _precondition(request, owner) + body = await _body(request, _FlagBody) + session_id = _clean_id(body.session_id, "session id") + state = _profile_state(request, owner) + record = _thread_mode._load_coordinator(state, session_id) + if record is None: + record = _thread_mode._new_coordinator_record(session_id, enabled=body.enabled) + else: + record["enabled"] = bool(body.enabled) + _thread_mode._save_coordinator(state, record) + if native_context(request) != owner: + raise NativeAPIError(412, "context_changed", "The native context changed; reconcile the outcome.") + return _response({"ok": True, "session_id": session_id, "enabled": bool(body.enabled)}, + owner, request_id) + + +@router.post("/register") +async def register(request: Request) -> Response: + owner = native_context(request) + request_id = _precondition(request, owner) + body = await _body(request, _RegisterBody) + coordinator_session_id = _clean_id(body.coordinator_session_id, "coordinator session id") + worker_session_id = _clean_id(body.worker_session_id, "worker session id") + state = _profile_state(request, owner) + record = _thread_mode._ensure_coordinator(state, coordinator_session_id) + if body.name in (record.get("threads") or {}): + raise NativeAPIError(409, "thread_name_conflict", + f"A thread named '{body.name}' already exists for this coordinator.") + entry = _thread_mode._new_thread_record(body.name, body.brief.strip(), "session") + entry["worker_session_id"] = worker_session_id + # The app created a real session and submitted the brief already: the + # worker is live from the plugin's point of view. + entry["status"] = "running" + entry["updated_at"] = time.time() + record["threads"][body.name] = entry + _thread_mode._save_coordinator(state, record) + state.set(_thread_mode.worker_key(worker_session_id), { + "coordinator_session_id": coordinator_session_id, + "thread": body.name, + }) + if native_context(request) != owner: + raise NativeAPIError(412, "context_changed", "The native context changed; reconcile the outcome.") + return _response({"ok": True, "thread": _thread_mode.public_thread(entry)}, owner, request_id) + + +@router.get("/roster") +async def roster(request: Request) -> Response: + owner = native_context(request) + request_id = _precondition(request, owner) + coordinator_session_id = _clean_id( + request.query_params.get("coordinator_session_id"), "coordinator session id") + state = _profile_state(request, owner) + record = _thread_mode._load_coordinator(state, coordinator_session_id) + if record is None: + raise NativeAPIError(404, "coordinator_not_found", + "No thread-mode record exists for this session.") + if native_context(request) != owner: + raise NativeAPIError(412, "context_changed", "The native context changed; reconcile the outcome.") + return _response({ + "coordinator_session_id": coordinator_session_id, + "enabled": bool(record.get("enabled")), + "updated_at": record.get("updated_at"), + "notes": list(record.get("notes") or []), + "threads": [_thread_mode.public_thread(entry) + for entry in (record.get("threads") or {}).values()], + }, owner, request_id) diff --git a/loopdy_plugin/registration.py b/loopdy_plugin/registration.py index 366440c..0319e83 100644 --- a/loopdy_plugin/registration.py +++ b/loopdy_plugin/registration.py @@ -211,6 +211,11 @@ def approval_requested(**kwargs): bridge=device_tool_bridge if legacy_device_tools else None, ) register_marketplace_publish_skill(ctx) + # Thread mode (coordinator/worker threads): four loopdy tools plus additive + # pre_llm_call / subagent_start / subagent_stop hooks. Registered next to + # the existing tool registration; never alters the existing hooks. + from .thread_mode import register as register_thread_mode + register_thread_mode(ctx) ctx.register_platform( name="loopdy", diff --git a/loopdy_plugin/thread_mode.py b/loopdy_plugin/thread_mode.py new file mode 100644 index 0000000..c37c65d --- /dev/null +++ b/loopdy_plugin/thread_mode.py @@ -0,0 +1,708 @@ +"""Loopdy thread mode: coordinator/worker threads (plugin side). + +Implements PROTOCOL.md: four ``loopdy``-toolset tools (thread_spawn, +thread_status, thread_collect, thread_note), the pre_llm_call injection for +coordinator/worker turns, and subagent_start/subagent_stop observers. All +state lives in Hermes PluginState (profile-scoped JSON KV) under +``threadmode:`` and ``threadmode:worker:`` keys. + +NO Hermes core changes. Workers are launched via ctx.subagent_lifecycle +(in-process subagents) or created out-of-turn by the Loopdy app as real +sessions (registered via the native REST surface). REST session.fork is +never used for workers. +""" +from __future__ import annotations + +import json +import logging +import re +import time +from enum import Enum +from typing import Any + +logger = logging.getLogger("hermes.plugins.loopdy.thread_mode") + +THREAD_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") +MAX_NOTES_PER_THREAD = 50 +MAX_NOTE_CHARS = 2000 +MAX_RESULT_CHARS = 20000 +MAX_BRIEF_CHARS = 4000 + +STATUS_VALUES = ( + "spawning", "running", "succeeded", "failed", + "cancelled", "interrupted", "unknown", +) +TERMINAL_STATUSES = {"succeeded", "failed", "cancelled", "interrupted"} + +#: Maps Hermes SubagentState values onto the thread-mode status vocabulary. +_SUBAGENT_STATUS_MAP = { + "PENDING": "spawning", + "STARTING": "spawning", + "RUNNING": "running", + "CANCEL_REQUESTED": "running", + "SUCCEEDED": "succeeded", + "FAILED": "failed", + "INTERRUPTED": "interrupted", + "CANCELLED": "cancelled", + "UNKNOWN": "unknown", +} + +#: Maps subagent_stop hook child_status values onto our vocabulary. +_CHILD_STATUS_MAP = { + "completed": "succeeded", + "interrupted": "interrupted", + "failed": "failed", +} + +#: Clean error returned when thread_spawn is called without a live turn. +NO_LIVE_TURN_ERROR = ( + "thread_spawn needs a live coordinator turn (no active Hermes parent " + "session). Create the worker from the Loopdy thread view instead." +) + +_RESULT_HEADING_RE = re.compile(r"^##\s+Result\s*$", re.MULTILINE) +_NEXT_SECTION_RE = re.compile(r"^#{1,2}\s", re.MULTILINE) + + +def coordinator_key(coordinator_session_id: str) -> str: + return f"threadmode:{coordinator_session_id}" + + +def worker_key(worker_session_id: str) -> str: + return f"threadmode:worker:{worker_session_id}" + + +def _utcnow() -> float: + return time.time() + + +def _new_coordinator_record(coordinator_session_id: str, enabled: bool = True) -> dict: + return { + "version": 1, + "enabled": bool(enabled), + "coordinator_session_id": coordinator_session_id, + "updated_at": _utcnow(), + "notes": [], + "threads": {}, + } + + +def _load_coordinator(state: Any, coordinator_session_id: str) -> dict | None: + if not isinstance(coordinator_session_id, str) or not coordinator_session_id: + return None + record = state.get(coordinator_key(coordinator_session_id)) + if not isinstance(record, dict): + return None + threads = record.get("threads") + if not isinstance(threads, dict): + record["threads"] = {} + if not isinstance(record.get("notes"), list): + record["notes"] = [] + return record + + +def _save_coordinator(state: Any, record: dict) -> None: + record["updated_at"] = _utcnow() + state.set(coordinator_key(record["coordinator_session_id"]), record) + + +def _ensure_coordinator(state: Any, coordinator_session_id: str) -> dict: + record = _load_coordinator(state, coordinator_session_id) + if record is None: + record = _new_coordinator_record(coordinator_session_id, enabled=True) + return record + + +def _new_thread_record(name: str, brief: str, kind: str) -> dict: + now = _utcnow() + return { + "name": name, + "brief": brief, + "kind": kind, + "status": "spawning", + "subagent_id": None, + "subagent_session_id": None, + "handle": None, + "worker_session_id": None, + "notes": [], + "result": None, + "created_at": now, + "updated_at": now, + } + + +def public_thread(entry: dict) -> dict: + """Roster projection: everything except the raw subagent handle dict.""" + ids = {} + for key in ("subagent_id", "subagent_session_id", "worker_session_id"): + value = entry.get(key) + if value: + ids[key] = value + return { + "name": entry.get("name"), + "kind": entry.get("kind"), + "status": entry.get("status"), + "brief": entry.get("brief"), + "ids": ids, + "notes": list(entry.get("notes") or []), + "result": entry.get("result"), + "updated_at": entry.get("updated_at"), + } + + +def parse_result_section(text: Any) -> str: + """Return the first ``## Result`` heading section; fallback = full text.""" + if not isinstance(text, str) or not text: + return "" + match = _RESULT_HEADING_RE.search(text) + if match is None: + return text + rest = text[match.end():] + following = _NEXT_SECTION_RE.search(rest) + section = rest[: following.start()] if following else rest + return section.strip() or text + + +def worker_discipline(name: str, brief: str, extra_context: str = "") -> str: + """The §8 worker discipline, injected via launch context / prompt.""" + lines = [ + f"You are worker thread `{name}` in a Loopdy thread-mode session. " + "Do ONLY the scoped brief below. Do not expand scope.", + "", + "## Your assignment", + brief, + ] + if extra_context and extra_context.strip(): + lines += ["", extra_context.strip()] + lines += [ + "", + "End your FINAL message with a `## Result` section containing the complete deliverable.", + "NEVER write to global/profile memory (`MEMORY.md`). Scratch notes go via the " + "`thread_note` tool (available in your toolset) — or just include them in your result.", + "Sibling threads exist and work in parallel; shared context for you is injected above. " + "Do not wait on them.", + ] + return "\n".join(lines) + + +def _tool_json(value: dict) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _map_subagent_state(state_value: Any) -> str: + name = state_value.name if isinstance(state_value, Enum) else str(state_value) + return _SUBAGENT_STATUS_MAP.get(name, "unknown") + + +# --------------------------------------------------------------------------- +# Tool handlers +# --------------------------------------------------------------------------- + +def _thread_spawn_handler(ctx: Any): + def handle(payload: Any, **kwargs: Any) -> str: + coordinator_session_id = str(kwargs.get("session_id") or "") + if not isinstance(payload, dict): + return _tool_json({"ok": False, "error": "thread_spawn expects a JSON object payload."}) + specs = payload.get("threads") + if not isinstance(specs, list) or not specs: + return _tool_json({"ok": False, "error": "thread_spawn needs at least one thread."}) + if not coordinator_session_id: + return _tool_json({"ok": False, "error": "thread_spawn needs a coordinator session."}) + state = ctx.state + record = _ensure_coordinator(state, coordinator_session_id) + results = [ + _spawn_one(ctx, state, record, coordinator_session_id, spec) + for spec in specs + ] + _save_coordinator(state, record) + return _tool_json({"ok": True, "threads": results}) + + return handle + + +def _spawn_one( + ctx: Any, state: Any, record: dict, coordinator_session_id: str, spec: Any, +) -> dict: + name = spec.get("name") if isinstance(spec, dict) else None + brief = spec.get("brief") if isinstance(spec, dict) else None + extra = spec.get("context") if isinstance(spec, dict) else None + if not isinstance(name, str) or THREAD_NAME_RE.fullmatch(name) is None: + return { + "name": name if isinstance(name, str) else None, + "ok": False, "kind": "subagent", + "error": f"Invalid thread name {name!r}: must match ^[a-z0-9][a-z0-9_-]{{0,63}}$.", + } + if name in record["threads"]: + return { + "name": name, "ok": False, "kind": "subagent", + "error": f"A thread named '{name}' already exists for this coordinator.", + } + if not isinstance(brief, str) or not brief.strip(): + return { + "name": name, "ok": False, "kind": "subagent", + "error": f"Thread '{name}' needs a non-empty brief.", + } + brief = brief.strip()[:MAX_BRIEF_CHARS] + + # Persist the spawning record BEFORE launch: this closes the race where + # the child's first pre_llm_call fires before the tool persists anything. + entry = _new_thread_record(name, brief, "subagent") + record["threads"][name] = entry + _save_coordinator(state, record) + + from agent.subagent_lifecycle import SubagentLaunchRequest, SubagentLifecycleError + + try: + request = SubagentLaunchRequest( + goal=brief, + context=worker_discipline(name, brief, extra if isinstance(extra, str) else ""), + role="leaf", + correlation_id=name, + metadata={"threadmode": coordinator_session_id, "thread": name}, + ) + handle = ctx.subagent_lifecycle.launch(request) + except SubagentLifecycleError as exc: + entry["status"] = "failed" + entry["updated_at"] = _utcnow() + _save_coordinator(state, record) + message = str(exc) + if "No active Hermes parent session" in message: + error = NO_LIVE_TURN_ERROR + elif "Duplicate correlation_id" in message: + error = f"A worker for thread '{name}' is already launching in this coordinator turn." + else: + error = f"thread_spawn failed for '{name}': {message[:400]}" + return {"name": name, "ok": False, "kind": "subagent", "error": error} + except Exception as exc: # never leak a traceback to the model + entry["status"] = "failed" + entry["updated_at"] = _utcnow() + _save_coordinator(state, record) + return { + "name": name, "ok": False, "kind": "subagent", + "error": f"thread_spawn failed for '{name}': {type(exc).__name__}.", + } + entry["subagent_id"] = handle.subagent_id + entry["handle"] = handle.to_dict() + entry["status"] = "running" + entry["updated_at"] = _utcnow() + _save_coordinator(state, record) + return {"name": name, "ok": True, "kind": "subagent", "subagent_id": handle.subagent_id} + + +def _refresh_subagent_status(ctx: Any, entry: dict) -> str | None: + """Best-effort refresh of a subagent-kind thread's status; None on failure.""" + handle_dict = entry.get("handle") + if not isinstance(handle_dict, dict): + return None + try: + from agent.subagent_lifecycle import SubagentHandle + handle = SubagentHandle.from_dict(handle_dict) + status = ctx.subagent_lifecycle.status(handle) + except Exception: + logger.warning("Thread-mode subagent status refresh failed", exc_info=True) + return None + return _map_subagent_state(getattr(status, "state", "UNKNOWN")) + + +def _thread_status_handler(ctx: Any): + def handle(payload: Any, **kwargs: Any) -> str: + coordinator_session_id = str(kwargs.get("session_id") or "") + record = _load_coordinator(ctx.state, coordinator_session_id) + if record is None: + return _tool_json({"ok": True, "threads": []}) + changed = False + for entry in record["threads"].values(): + if ( + entry.get("kind") == "subagent" + and entry.get("status") not in TERMINAL_STATUSES + ): + refreshed = _refresh_subagent_status(ctx, entry) + if refreshed is not None and refreshed != entry.get("status"): + entry["status"] = refreshed + entry["updated_at"] = _utcnow() + changed = True + if changed: + _save_coordinator(ctx.state, record) + return _tool_json( + {"ok": True, "threads": [public_thread(e) for e in record["threads"].values()]} + ) + + return handle + + +def _read_worker_session_text(ctx: Any, worker_session_id: Any) -> str | None: + """Best-effort control-plane read of a session-kind worker's last text. + + Returns the last assistant message's text, or None when unavailable. + """ + if not isinstance(worker_session_id, str) or not worker_session_id: + return None + try: + from hermes_state import SessionDB + from hermes_cli.profiles import get_profile_dir + profile = str(getattr(ctx, "profile_name", "default") or "default") + db = SessionDB(db_path=get_profile_dir(profile) / "state.db", read_only=True) + try: + messages = db.get_messages(worker_session_id, latest=True, limit=50) + finally: + db.close() + except Exception: + logger.warning("Thread-mode worker session read failed", exc_info=True) + return None + for message in reversed(messages or []): + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + content = message.get("content") + if isinstance(content, str) and content.strip(): + return content + return None + + +def _thread_collect_handler(ctx: Any): + def handle(payload: Any, **kwargs: Any) -> str: + coordinator_session_id = str(kwargs.get("session_id") or "") + name = payload.get("thread_name") if isinstance(payload, dict) else None + record = _load_coordinator(ctx.state, coordinator_session_id) + entry = record["threads"].get(name) if record and isinstance(name, str) else None + if entry is None: + return _tool_json({"collected": False, "error": f"Unknown thread '{name}'."}) + if entry.get("kind") == "session": + text = _read_worker_session_text(ctx, entry.get("worker_session_id")) + if text is None: + return _tool_json({ + "collected": False, + "thread_name": name, + "worker_session_id": entry.get("worker_session_id"), + "hint": "Read the worker session directly.", + }) + result = parse_result_section(text)[:MAX_RESULT_CHARS] + entry["result"] = result + entry["updated_at"] = _utcnow() + _save_coordinator(ctx.state, record) + return _tool_json({"collected": True, "thread_name": name, "result": result}) + handle_dict = entry.get("handle") + if not isinstance(handle_dict, dict): + return _tool_json({"collected": False, "thread_name": name, + "status": entry.get("status"), + "error": "Subagent handle is missing."}) + try: + from agent.subagent_lifecycle import SubagentHandle + handle = SubagentHandle.from_dict(handle_dict) + outcome = ctx.subagent_lifecycle.result(handle) + except Exception: + logger.warning("Thread-mode subagent result read failed", exc_info=True) + return _tool_json({"collected": False, "thread_name": name, + "status": entry.get("status")}) + if not getattr(outcome, "ready", False): + return _tool_json({"collected": False, "thread_name": name, + "status": entry.get("status")}) + # Never block: no wait() with a timeout here. + summary = getattr(outcome, "summary", None) + result = parse_result_section(summary)[:MAX_RESULT_CHARS] + entry["result"] = result + entry["updated_at"] = _utcnow() + _save_coordinator(ctx.state, record) + return _tool_json({"collected": True, "thread_name": name, "result": result}) + + return handle + + +def _thread_note_handler(ctx: Any): + def handle(payload: Any, **kwargs: Any) -> str: + coordinator_session_id = str(kwargs.get("session_id") or "") + if not isinstance(payload, dict): + return _tool_json({"ok": False, "error": "thread_note expects a JSON object payload."}) + note = payload.get("note") + if not isinstance(note, str) or not note.strip(): + return _tool_json({"ok": False, "error": "thread_note needs a non-empty note."}) + note = note.strip()[:MAX_NOTE_CHARS] + name = payload.get("thread_name") + record = _ensure_coordinator(ctx.state, coordinator_session_id) + if name is None: + notes = record.setdefault("notes", []) + else: + entry = record["threads"].get(name) if isinstance(name, str) else None + if entry is None: + return _tool_json({"ok": False, "error": f"Unknown thread '{name}'."}) + notes = entry.setdefault("notes", []) + notes.append(note) + del notes[: max(0, len(notes) - MAX_NOTES_PER_THREAD)] + _save_coordinator(ctx.state, record) + return _tool_json({"ok": True, "note_count": len(notes)}) + + return handle + + +# --------------------------------------------------------------------------- +# pre_llm_call injection +# --------------------------------------------------------------------------- + +def _coordinator_section(record: dict) -> str: + threads = record.get("threads") or {} + rows = ["| name | kind | status | brief |", "| --- | --- | --- | --- |"] + for entry in threads.values(): + rows.append( + f"| {entry.get('name')} | {entry.get('kind')} | {entry.get('status')} " + f"| {(entry.get('brief') or '')[:120]} |" + ) + roster = "\n".join(rows) if threads else "No threads yet." + return ( + "## Thread mode — coordinator\n" + "You are the coordinator of a Loopdy thread-mode session. " + f"{len(threads)} worker thread(s) run in parallel.\n" + f"{roster}\n" + "Delegate scoped work with `thread_spawn` while you have a live turn; the " + "Loopdy thread view can also launch session-backed workers out-of-turn. " + "Review progress with `thread_status` and gather finished work with " + "`thread_collect`; YOU assemble the final answer — never ask a worker to assemble it.\n" + "Memory discipline: ONLY you write durable profile memory (`MEMORY.md`). " + "Workers never do; all cross-thread scratch lives in thread-mode plugin state " + "and is injected into turns." + ) + + +def _worker_section(entry: dict, siblings: list[dict]) -> str: + notes = entry.get("notes") or [] + sibling_lines = [ + f"- {s.get('name')}: {s.get('status')}" + for s in siblings if s.get("name") != entry.get("name") + ] + lines = [ + "## Your thread assignment", + f"You are worker thread `{entry.get('name')}` in a Loopdy thread-mode session.", + "", + "Scoped brief:", + str(entry.get("brief") or ""), + "", + "Scratch notes for this thread:", + ] + lines += [f"- {note}" for note in notes] if notes else ["(none)"] + lines += ["", "Sibling threads (names + statuses only):"] + lines += sibling_lines if sibling_lines else ["(none)"] + lines += ["", worker_discipline(str(entry.get("name")), str(entry.get("brief") or ""))] + return "\n".join(lines) + + +def _resolve_worker_entry(record: dict, worker_session_id: str) -> dict | None: + """Match a worker turn's session id to its thread record.""" + threads = record.get("threads") or {} + for entry in threads.values(): + if entry.get("subagent_session_id") == worker_session_id: + return entry + # Fallback: the single spawning/running thread for that coordinator. + candidates = [ + entry for entry in threads.values() + if entry.get("status") in ("spawning", "running") + ] + if len(candidates) == 1: + return candidates[0] + return None + + +def pre_llm_call(state: Any, **payload: Any) -> dict | None: + """Additive thread-mode context injection. Never breaks the turn: fails silent.""" + try: + session_id = str(payload.get("session_id") or "") + parent_session_id = str(payload.get("parent_session_id") or "") + if not session_id: + return None + if parent_session_id: + # Subagent-kind worker turn. + record = _load_coordinator(state, parent_session_id) + if record is None or not record.get("enabled"): + return None + entry = _resolve_worker_entry(record, session_id) + if entry is None: + return None + siblings = list((record.get("threads") or {}).values()) + return {"context": _worker_section(entry, siblings)} + record = _load_coordinator(state, session_id) + if record is not None and record.get("enabled"): + return {"context": _coordinator_section(record)} + index = state.get(worker_key(session_id)) + if isinstance(index, dict): + coordinator_session_id = index.get("coordinator_session_id") + thread_name = index.get("thread") + record = _load_coordinator(state, coordinator_session_id) + if record is not None and record.get("enabled"): + entry = (record.get("threads") or {}).get(thread_name) + if entry is not None: + siblings = list((record.get("threads") or {}).values()) + return {"context": _worker_section(entry, siblings)} + return None + except Exception: + logger.warning("Thread-mode pre_llm_call injection failed", exc_info=True) + return None + + +# --------------------------------------------------------------------------- +# subagent_start / subagent_stop observers +# --------------------------------------------------------------------------- + +def subagent_start(state: Any, **payload: Any) -> None: + """Record child_session_id and write the worker index at subagent_start.""" + try: + parent_session_id = str(payload.get("parent_session_id") or "") + child_subagent_id = str(payload.get("child_subagent_id") or "") + child_session_id = str(payload.get("child_session_id") or "") + if not parent_session_id or not child_subagent_id or not child_session_id: + return + record = _load_coordinator(state, parent_session_id) + if record is None or not record.get("enabled"): + return + for name, entry in (record.get("threads") or {}).items(): + if entry.get("kind") == "subagent" and entry.get("subagent_id") == child_subagent_id: + entry["subagent_session_id"] = child_session_id + if entry.get("status") not in TERMINAL_STATUSES: + entry["status"] = "running" + entry["updated_at"] = _utcnow() + _save_coordinator(state, record) + state.set(worker_key(child_session_id), { + "coordinator_session_id": parent_session_id, + "thread": name, + }) + return + except Exception: + logger.warning("Thread-mode subagent_start observation failed", exc_info=True) + + +def subagent_stop(state: Any, **payload: Any) -> None: + """Map the child's terminal status onto the thread record.""" + try: + parent_session_id = str(payload.get("parent_session_id") or "") + child_session_id = str(payload.get("child_session_id") or "") + if not parent_session_id or not child_session_id: + return + record = _load_coordinator(state, parent_session_id) + if record is None or not record.get("enabled"): + return + mapped = _CHILD_STATUS_MAP.get(str(payload.get("child_status") or "")) + if mapped is None: + return + for entry in (record.get("threads") or {}).values(): + if ( + entry.get("kind") == "subagent" + and entry.get("subagent_session_id") == child_session_id + ): + entry["status"] = mapped + entry["updated_at"] = _utcnow() + _save_coordinator(state, record) + return + except Exception: + logger.warning("Thread-mode subagent_stop observation failed", exc_info=True) + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + +THREAD_TOOLS = ( + ( + "thread_spawn", + "Launch one or more parallel worker threads (Hermes in-process subagents) from a live " + "coordinator turn. Each thread needs a unique name matching ^[a-z0-9][a-z0-9_-]{0,63}$ " + "and a scoped brief. The scoped brief becomes the worker's assignment; thread_spawn " + "needs a live coordinator turn. Requires thread mode enabled for this conversation.", + { + "type": "object", + "properties": { + "threads": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "properties": { + "name": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$"}, + "brief": {"type": "string", "minLength": 1, "maxLength": 4000}, + "context": {"type": "string", "maxLength": 4000}, + }, + "required": ["name", "brief"], + "additionalProperties": False, + }, + }, + }, + "required": ["threads"], + "additionalProperties": False, + }, + ), + ( + "thread_status", + "Report the current roster of thread-mode worker threads: name, kind, status, brief, " + "ids, and timestamps. Subagent-kind statuses refresh from the Hermes subagent lifecycle.", + { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + ), + ( + "thread_collect", + "Collect a worker thread's finished deliverable. Parses the worker's `## Result` " + "section (falls back to the full final text). Never blocks. For session-backed " + "threads the result comes from the worker session's recent messages when available.", + { + "type": "object", + "properties": { + "thread_name": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$"}, + }, + "required": ["thread_name"], + "additionalProperties": False, + }, + ), + ( + "thread_note", + "Append a scratch note to a worker thread (max 50 notes, 2000 chars each). Omit " + "thread_name to write a coordinator-level note. Only the coordinator writes " + "durable profile memory; thread notes are cross-thread scratch.", + { + "type": "object", + "properties": { + "thread_name": {"type": "string", "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$"}, + "note": {"type": "string", "minLength": 1, "maxLength": 2000}, + }, + "required": ["note"], + "additionalProperties": False, + }, + ), +) + + +def register(ctx: Any) -> None: + """Register thread-mode tools and additive hooks. Closes over ctx.""" + handlers = { + "thread_spawn": _thread_spawn_handler(ctx), + "thread_status": _thread_status_handler(ctx), + "thread_collect": _thread_collect_handler(ctx), + "thread_note": _thread_note_handler(ctx), + } + for name, description, parameters in THREAD_TOOLS: + ctx.register_tool( + name=name, + toolset="loopdy", + schema={"name": name, "description": description, "parameters": parameters}, + handler=handlers[name], + ) + # Additive hooks: registered alongside the existing ones; fail silent. + # ctx.state resolves lazily at fire time so registration never assumes + # a state facade the host may not provide. + ctx.register_hook("pre_llm_call", lambda **payload: pre_llm_call(ctx.state, **payload)) + ctx.register_hook("subagent_start", lambda **payload: subagent_start(ctx.state, **payload)) + ctx.register_hook("subagent_stop", lambda **payload: subagent_stop(ctx.state, **payload)) + # Expose the profile's PluginState to the out-of-turn REST surface + # (native_threads scans sys.modules for the loader-owned module). + from .native_threads import register_state_resolver + register_state_resolver(str(getattr(ctx, "profile_name", "default") or "default"), + _LazyState(ctx)) + + +class _LazyState: + """Defers to ctx.state so registration works without an eager facade.""" + + def __init__(self, ctx: Any) -> None: + self._ctx = ctx + + def get(self, key: str, default: Any = None) -> Any: + return self._ctx.state.get(key, default) + + def set(self, key: str, value: Any) -> None: + self._ctx.state.set(key, value) diff --git a/plugin.yaml b/plugin.yaml index 97794aa..4c0a912 100644 --- a/plugin.yaml +++ b/plugin.yaml @@ -69,6 +69,10 @@ provides_tools: - loopdy_render_card_template - loopdy_marketplace_prepare_upload - loopdy_react_to_message + - thread_spawn + - thread_status + - thread_collect + - thread_note platforms: - linux - macos diff --git a/tests/test_thread_mode.py b/tests/test_thread_mode.py new file mode 100644 index 0000000..fbc88b0 --- /dev/null +++ b/tests/test_thread_mode.py @@ -0,0 +1,724 @@ +"""Thread-mode plugin contract tests: tools, hooks, REST, injection. + +Uses unittest discovery (``python -m unittest discover -s tests``), matching +the repo's CI. The Hermes ``agent.subagent_lifecycle`` module is faked in +sys.modules; the dashboard REST tests reuse the real Hermes auth middleware +with a fixture provider, mirroring tests/test_native_api.py. +""" +from __future__ import annotations + +import json +import os +import sys +import tempfile +import time +import unittest +import uuid +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] + + +# --------------------------------------------------------------------------- +# Hermes lifecycle types (real code, verified at ~/workspace/hermes-agent). +# The handlers lazy-import these inside the turn; the tests import the same +# module so construction matches the real dataclass shapes. +# --------------------------------------------------------------------------- +from agent.subagent_lifecycle import ( # noqa: E402 + SubagentHandle, + SubagentLaunchRequest, + SubagentLifecycleError, + SubagentResult, + SubagentState, + SubagentStatus, +) + +from loopdy_plugin import thread_mode # noqa: E402 + + +def _handle(subagent_id="sub-1", parent="coord-1", correlation="worker-a"): + return SubagentHandle(1, subagent_id, parent, correlation, time.time(), + None, None, "leaf", 1, "capability") + + +class FakeState: + def __init__(self): + self._data = {} + + def get(self, key, default=None): + return self._data.get(key, default) + + def set(self, key, value): + self._data[key] = value + + +class FakeLifecycle: + def __init__(self): + self.launched = [] + self.launch_behavior = None + self.statuses = {} + self.results = {} + + def launch(self, request): + self.launched.append(request) + if self.launch_behavior is not None: + return self.launch_behavior(request) + return _handle(correlation=request.correlation_id) + + def status(self, handle): + return self.statuses.get(handle.subagent_id, + SubagentStatus(handle, SubagentState.RUNNING, time.time())) + + def result(self, handle): + return self.results.get(handle.subagent_id, + SubagentResult(handle, SubagentState.RUNNING, False)) + + +class FakeCtx: + profile_name = "default" + + def __init__(self): + self.tools = {} + self.schemas = {} + self.toolsets = {} + self.hooks = {} + self.state = FakeState() + self.subagent_lifecycle = FakeLifecycle() + + def register_tool(self, *, name, handler, schema, toolset=None, **_kwargs): + self.tools[name] = handler + self.schemas[name] = schema + self.toolsets[name] = toolset + + def register_hook(self, name, callback): + self.hooks.setdefault(name, []).append(callback) + + +COORD = "coord-session-1" + + +def _spawn_payload(*specs): + return {"threads": [{"name": name, "brief": brief, **extra} + for name, brief, extra in specs]} + + +class ThreadToolRegistrationTests(unittest.TestCase): + def test_four_tools_registered_in_loopdy_toolset(self): + ctx = FakeCtx() + thread_mode.register(ctx) + for name in ("thread_spawn", "thread_status", "thread_collect", "thread_note"): + self.assertIn(name, ctx.tools) + self.assertEqual(ctx.toolsets[name], "loopdy") + schema = ctx.schemas[name] + self.assertEqual(schema["name"], name) + self.assertIn("parameters", schema) + + def test_additive_hooks_registered(self): + ctx = FakeCtx() + thread_mode.register(ctx) + for hook in ("pre_llm_call", "subagent_start", "subagent_stop"): + self.assertIn(hook, ctx.hooks) + self.assertEqual(len(ctx.hooks[hook]), 1) + + def test_plugin_yaml_provides_thread_tools(self): + import yaml + manifest = yaml.safe_load((ROOT / "plugin.yaml").read_text(encoding="utf-8")) + provides = manifest["provides_tools"] + for name in ("thread_spawn", "thread_status", "thread_collect", "thread_note"): + self.assertIn(name, provides) + for hook in ("pre_llm_call", "subagent_start", "subagent_stop"): + self.assertIn(hook, manifest["provides_hooks"]) + + def test_dashboard_mounts_threads_router(self): + source = (ROOT / "dashboard" / "plugin_api.py").read_text(encoding="utf-8") + self.assertIn("native_threads_router", source) + self.assertIn("include_router(native_threads_router)", source) + + +class ThreadNameValidationTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + + def _spawn(self, *specs): + return json.loads(self.spawn(_spawn_payload(*specs), session_id=COORD)) + + def test_rejects_bad_names_per_thread_and_continues(self): + result = self._spawn( + ("Bad Name", "brief", {}), + ("-leading", "brief", {}), + ("ok-name_1", "brief", {}), + ("x" * 65, "brief", {}), + ("", "brief", {}), + ) + by_name = {item["name"]: item for item in result["threads"]} + self.assertFalse(by_name["Bad Name"]["ok"]) + self.assertFalse(by_name["-leading"]["ok"]) + self.assertFalse(by_name["x" * 65]["ok"]) + self.assertFalse(by_name[None]["ok"] if None in by_name else by_name[""]["ok"]) + self.assertTrue(by_name["ok-name_1"]["ok"]) + for item in result["threads"]: + if not item["ok"]: + self.assertIn("error", item) + self.assertIn("^[a-z0-9][a-z0-9_-]{0,63}$", item["error"]) + + def test_rejects_duplicate_name_per_coordinator(self): + self._spawn(("dup", "first", {})) + result = self._spawn(("dup", "second", {})) + self.assertFalse(result["threads"][0]["ok"]) + self.assertIn("already exists", result["threads"][0]["error"]) + + def test_rejects_empty_brief(self): + result = self._spawn(("nobrieef", " ", {})) + self.assertFalse(result["threads"][0]["ok"]) + self.assertIn("brief", result["threads"][0]["error"]) + + def test_rejects_empty_thread_list(self): + result = json.loads(self.spawn({"threads": []}, session_id=COORD)) + self.assertFalse(result["ok"]) + + +class ThreadSpawnLifecycleTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + + def test_spawning_record_persisted_before_launch(self): + seen = {} + + def launch(request): + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + seen["status_at_launch"] = record["threads"]["early"]["status"] + seen["request"] = request + return _handle(correlation=request.correlation_id) + + self.ctx.subagent_lifecycle.launch_behavior = launch + result = json.loads(self.spawn(_spawn_payload(("early", "do things", {})), + session_id=COORD)) + self.assertEqual(seen["status_at_launch"], "spawning") + self.assertTrue(result["threads"][0]["ok"]) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["early"]["status"], "running") + self.assertEqual(record["threads"]["early"]["subagent_id"], "sub-1") + self.assertIn("handle", record["threads"]["early"]) + + def test_launch_request_carries_discipline_goal_role_metadata(self): + self.spawn(_spawn_payload(("worker-a", "scope the brief", {"context": "extra ctx"})), + session_id=COORD) + request = self.ctx.subagent_lifecycle.launched[0] + self.assertEqual(request.goal, "scope the brief") + self.assertEqual(request.role, "leaf") + self.assertEqual(request.correlation_id, "worker-a") + self.assertEqual(request.metadata, {"threadmode": COORD, "thread": "worker-a"}) + self.assertIn("worker-a", request.context) + self.assertIn("scope the brief", request.context) + self.assertIn("extra ctx", request.context) + self.assertIn("## Result", request.context) + self.assertIn("MEMORY.md", request.context) + + def test_no_live_turn_returns_clean_error_without_traceback(self): + def launch(_request): + raise SubagentLifecycleError("No active Hermes parent session is available.") + + self.ctx.subagent_lifecycle.launch_behavior = launch + result = json.loads(self.spawn(_spawn_payload(("ghost", "brief", {})), + session_id=COORD)) + item = result["threads"][0] + self.assertFalse(item["ok"]) + self.assertEqual(item["error"], thread_mode.NO_LIVE_TURN_ERROR) + self.assertIn("Loopdy thread view", item["error"]) + self.assertNotIn("Traceback", json.dumps(result)) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["ghost"]["status"], "failed") + + def test_duplicate_correlation_id_is_clear_error(self): + def launch(_request): + raise SubagentLifecycleError("Duplicate correlation_id for this parent session.") + + self.ctx.subagent_lifecycle.launch_behavior = launch + result = json.loads(self.spawn(_spawn_payload(("twice", "brief", {})), + session_id=COORD)) + self.assertFalse(result["threads"][0]["ok"]) + self.assertIn("already launching", result["threads"][0]["error"]) + + def test_unexpected_launch_error_never_leaks_traceback(self): + def launch(_request): + raise RuntimeError("boom\nTraceback (most recent call last): ...") + + self.ctx.subagent_lifecycle.launch_behavior = launch + result = json.loads(self.spawn(_spawn_payload(("oops", "brief", {})), + session_id=COORD)) + self.assertFalse(result["threads"][0]["ok"]) + self.assertNotIn("Traceback", result["threads"][0]["error"]) + + +class ThreadStatusTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + self.status = self.ctx.tools["thread_status"] + self.spawn(_spawn_payload(("worker-a", "brief a", {}), ("worker-b", "brief b", {})), + session_id=COORD) + + def test_refreshes_subagent_status_and_persists(self): + self.ctx.subagent_lifecycle.statuses["sub-1"] = SubagentStatus( + _handle("sub-1"), SubagentState.SUCCEEDED, time.time()) + result = json.loads(self.status({}, session_id=COORD)) + by_name = {item["name"]: item for item in result["threads"]} + self.assertEqual(by_name["worker-a"]["status"], "succeeded") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["status"], "succeeded") + + def test_session_kind_reports_stored_status(self): + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + entry = thread_mode._new_thread_record("sess-w", "brief", "session") + entry["status"] = "running" + entry["worker_session_id"] = "worker-session-9" + record["threads"]["sess-w"] = entry + self.ctx.state.set(thread_mode.coordinator_key(COORD), record) + result = json.loads(self.status({}, session_id=COORD)) + by_name = {item["name"]: item for item in result["threads"]} + self.assertEqual(by_name["sess-w"]["status"], "running") + self.assertEqual(by_name["sess-w"]["kind"], "session") + self.assertEqual(by_name["sess-w"]["ids"]["worker_session_id"], "worker-session-9") + self.assertNotIn("handle", json.dumps(by_name["sess-w"])) + + def test_unknown_coordinator_returns_empty_roster(self): + result = json.loads(self.status({}, session_id="nope")) + self.assertEqual(result["threads"], []) + + +class ThreadCollectTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + self.collect = self.ctx.tools["thread_collect"] + self.spawn(_spawn_payload(("worker-a", "brief", {})), session_id=COORD) + + def test_not_ready_reports_status(self): + result = json.loads(self.collect({"thread_name": "worker-a"}, session_id=COORD)) + self.assertFalse(result["collected"]) + self.assertEqual(result["status"], "running") + + def test_ready_parses_result_section(self): + self.ctx.subagent_lifecycle.results["sub-1"] = SubagentResult( + _handle("sub-1"), SubagentState.SUCCEEDED, True, + summary="Some chatter\n\n## Result\n\nThe deliverable.\n\n## Notes\nscratch") + result = json.loads(self.collect({"thread_name": "worker-a"}, session_id=COORD)) + self.assertTrue(result["collected"]) + self.assertEqual(result["result"], "The deliverable.") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["result"], "The deliverable.") + + def test_ready_without_result_heading_falls_back_to_full_text(self): + self.ctx.subagent_lifecycle.results["sub-1"] = SubagentResult( + _handle("sub-1"), SubagentState.SUCCEEDED, True, summary="just prose") + result = json.loads(self.collect({"thread_name": "worker-a"}, session_id=COORD)) + self.assertTrue(result["collected"]) + self.assertEqual(result["result"], "just prose") + + def test_session_kind_without_readable_session_returns_hint(self): + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + entry = thread_mode._new_thread_record("sess-w", "brief", "session") + entry["worker_session_id"] = "missing-session" + record["threads"]["sess-w"] = entry + self.ctx.state.set(thread_mode.coordinator_key(COORD), record) + with patch("hermes_state.SessionDB", side_effect=RuntimeError("no db")): + result = json.loads(self.collect({"thread_name": "sess-w"}, session_id=COORD)) + self.assertFalse(result["collected"]) + self.assertEqual(result["worker_session_id"], "missing-session") + self.assertIn("hint", result) + + def test_unknown_thread(self): + result = json.loads(self.collect({"thread_name": "nope"}, session_id=COORD)) + self.assertFalse(result["collected"]) + self.assertIn("error", result) + + +class ParseResultSectionTests(unittest.TestCase): + def test_first_result_section_only(self): + text = "# Title\n\n## Result\n\nLine one.\nLine two.\n\n## Result\n\nSecond.\n" + self.assertEqual(thread_mode.parse_result_section(text), "Line one.\nLine two.") + + def test_stops_before_next_h2(self): + text = "## Result\n\ndone\n\n## Appendix\n\nnope" + self.assertEqual(thread_mode.parse_result_section(text), "done") + + def test_missing_heading_returns_full_text(self): + self.assertEqual(thread_mode.parse_result_section("hello"), "hello") + + def test_non_string_returns_empty(self): + self.assertEqual(thread_mode.parse_result_section(None), "") + + +class ThreadNoteTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.note = self.ctx.tools["thread_note"] + self.spawn = self.ctx.tools["thread_spawn"] + self.spawn(_spawn_payload(("worker-a", "brief", {})), session_id=COORD) + + def test_thread_and_coordinator_notes(self): + first = json.loads(self.note({"thread_name": "worker-a", "note": "hello"}, session_id=COORD)) + self.assertTrue(first["ok"]) + self.assertEqual(first["note_count"], 1) + second = json.loads(self.note({"note": "coordinator scratch"}, session_id=COORD)) + self.assertTrue(second["ok"]) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["notes"], ["hello"]) + self.assertEqual(record["notes"], ["coordinator scratch"]) + + def test_notes_capped_at_fifty(self): + for index in range(60): + self.note({"thread_name": "worker-a", "note": f"n{index}"}, session_id=COORD) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + notes = record["threads"]["worker-a"]["notes"] + self.assertEqual(len(notes), 50) + self.assertEqual(notes[0], "n10") + self.assertEqual(notes[-1], "n59") + + def test_note_truncated_to_2000_chars(self): + self.note({"thread_name": "worker-a", "note": "x" * 3000}, session_id=COORD) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(len(record["threads"]["worker-a"]["notes"][0]), 2000) + + def test_unknown_thread_is_error(self): + result = json.loads(self.note({"thread_name": "nope", "note": "x"}, session_id=COORD)) + self.assertFalse(result["ok"]) + + def test_empty_note_is_error(self): + result = json.loads(self.note({"thread_name": "worker-a", "note": " "}, session_id=COORD)) + self.assertFalse(result["ok"]) + + +class PreLlmCallInjectionTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + self.note = self.ctx.tools["thread_note"] + + def _inject(self, **payload): + callbacks = self.ctx.hooks["pre_llm_call"] + self.assertEqual(len(callbacks), 1) + return callbacks[0](**payload) + + def test_coordinator_turn_injects_roster_and_memory_discipline(self): + self.spawn(_spawn_payload(("worker-a", "brief a", {}), ("worker-b", "brief b", {})), + session_id=COORD) + result = self._inject(session_id=COORD, parent_session_id="") + self.assertIsNotNone(result) + context = result["context"] + self.assertIn("## Thread mode — coordinator", context) + self.assertIn("worker-a", context) + self.assertIn("worker-b", context) + self.assertIn("thread_spawn", context) + self.assertIn("thread_status", context) + self.assertIn("thread_collect", context) + self.assertIn("never ask a worker to assemble", context) + self.assertIn("ONLY you write durable profile memory", context) + + def test_non_thread_session_injects_nothing(self): + self.assertIsNone(self._inject(session_id="plain", parent_session_id="")) + + def test_disabled_flag_injects_nothing(self): + self.spawn(_spawn_payload(("worker-a", "brief", {})), session_id=COORD) + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + record["enabled"] = False + self.ctx.state.set(thread_mode.coordinator_key(COORD), record) + self.assertIsNone(self._inject(session_id=COORD, parent_session_id="")) + + def test_subagent_worker_turn_injects_assignment(self): + self.spawn(_spawn_payload(("worker-a", "do the thing", {})), session_id=COORD) + self.note({"thread_name": "worker-a", "note": "remember x"}, session_id=COORD) + # subagent_start records the child session id + self.ctx.hooks["subagent_start"][0]( + parent_session_id=COORD, child_subagent_id="sub-1", child_session_id="child-9") + result = self._inject(session_id="child-9", parent_session_id=COORD) + self.assertIsNotNone(result) + context = result["context"] + self.assertIn("## Your thread assignment", context) + self.assertIn("worker-a", context) + self.assertIn("do the thing", context) + self.assertIn("remember x", context) + self.assertIn("## Result", context) + self.assertIn("MEMORY.md", context) + self.assertIn("Do not wait on them", context) + + def test_session_kind_worker_resolves_via_index(self): + record = thread_mode._ensure_coordinator(self.ctx.state, COORD) + entry = thread_mode._new_thread_record("sess-w", "session brief", "session") + entry["worker_session_id"] = "worker-sess-1" + entry["status"] = "running" + record["threads"]["sess-w"] = entry + self.ctx.state.set(thread_mode.coordinator_key(COORD), record) + self.ctx.state.set(thread_mode.worker_key("worker-sess-1"), + {"coordinator_session_id": COORD, "thread": "sess-w"}) + result = self._inject(session_id="worker-sess-1", parent_session_id="") + self.assertIsNotNone(result) + self.assertIn("## Your thread assignment", result["context"]) + self.assertIn("sess-w", result["context"]) + + def test_worker_turn_without_match_injects_nothing(self): + result = self._inject(session_id="stray", parent_session_id=COORD) + self.assertIsNone(result) + + def test_injection_fails_silent(self): + class BrokenState: + def get(self, *_args, **_kwargs): + raise RuntimeError("broken") + + def set(self, *_args, **_kwargs): + raise RuntimeError("broken") + + broken = FakeCtx() + broken.state = BrokenState() + thread_mode.register(broken) + callback = broken.hooks["pre_llm_call"][0] + self.assertIsNone(callback(session_id=COORD, parent_session_id="")) + + +class SubagentHookTests(unittest.TestCase): + def setUp(self): + self.ctx = FakeCtx() + thread_mode.register(self.ctx) + self.spawn = self.ctx.tools["thread_spawn"] + self.spawn(_spawn_payload(("worker-a", "brief", {})), session_id=COORD) + + def _hooks(self, name): + callbacks = self.ctx.hooks[name] + self.assertEqual(len(callbacks), 1) + return callbacks[0] + + def test_subagent_start_records_child_session_and_index(self): + self._hooks("subagent_start")( + parent_session_id=COORD, child_subagent_id="sub-1", child_session_id="child-9") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + entry = record["threads"]["worker-a"] + self.assertEqual(entry["subagent_session_id"], "child-9") + self.assertEqual(entry["status"], "running") + index = self.ctx.state.get(thread_mode.worker_key("child-9")) + self.assertEqual(index, {"coordinator_session_id": COORD, "thread": "worker-a"}) + + def test_subagent_start_ignores_unknown_child(self): + self._hooks("subagent_start")( + parent_session_id=COORD, child_subagent_id="nope", child_session_id="child-x") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertIsNone(record["threads"]["worker-a"]["subagent_session_id"]) + self.assertIsNone(self.ctx.state.get(thread_mode.worker_key("child-x"))) + + def test_subagent_stop_maps_terminal_status(self): + start = self._hooks("subagent_start") + stop = self._hooks("subagent_stop") + start(parent_session_id=COORD, child_subagent_id="sub-1", child_session_id="child-9") + stop(parent_session_id=COORD, child_session_id="child-9", child_status="completed") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["status"], "succeeded") + + def test_subagent_stop_maps_failed_and_interrupted(self): + start = self._hooks("subagent_start") + stop = self._hooks("subagent_stop") + start(parent_session_id=COORD, child_subagent_id="sub-1", child_session_id="child-9") + stop(parent_session_id=COORD, child_session_id="child-9", child_status="failed") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["status"], "failed") + stop(parent_session_id=COORD, child_session_id="child-9", child_status="interrupted") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["status"], "interrupted") + + def test_subagent_stop_unknown_status_leaves_record(self): + start = self._hooks("subagent_start") + stop = self._hooks("subagent_stop") + start(parent_session_id=COORD, child_subagent_id="sub-1", child_session_id="child-9") + stop(parent_session_id=COORD, child_session_id="child-9", child_status="weird") + record = self.ctx.state.get(thread_mode.coordinator_key(COORD)) + self.assertEqual(record["threads"]["worker-a"]["status"], "running") + + def test_hooks_fail_silent(self): + broken = FakeCtx() + + class BrokenState(FakeState): + def get(self, *_args, **_kwargs): + raise RuntimeError("broken") + + broken.state = BrokenState() + thread_mode.register(broken) + broken.hooks["subagent_start"][0](parent_session_id=COORD) + broken.hooks["subagent_stop"][0](parent_session_id=COORD) + + +class WorkerDisciplineTests(unittest.TestCase): + def test_discipline_contains_all_required_elements(self): + text = thread_mode.worker_discipline("worker-a", "the brief", "extra") + self.assertIn("worker-a", text) + self.assertIn("the brief", text) + self.assertIn("extra", text) + self.assertIn("Do ONLY", text) + self.assertIn("## Result", text) + self.assertIn("MEMORY.md", text) + self.assertIn("thread_note", text) + self.assertIn("Do not wait on them", text) + + +# --------------------------------------------------------------------------- +# REST tests (real Hermes auth middleware, fake plugin state) +# --------------------------------------------------------------------------- + +class NativeThreadsRestTests(unittest.TestCase): + PREFIX = "/api/plugins/loopdy/native/threads" + + @classmethod + def setUpClass(cls): + from hermes_cli.dashboard_auth.base import DashboardAuthProvider, Session + from hermes_cli.dashboard_auth.middleware import gated_auth_middleware + from hermes_cli.dashboard_auth.registry import register_provider, unregister_global_provider + from fastapi import FastAPI + from fastapi.testclient import TestClient + from loopdy_plugin import native_threads, native_api + + class FixtureProvider(DashboardAuthProvider): + name = "thread-mode-fixture" + display_name = "Thread mode fixture" + + def __init__(self): + self.alice = Session("alice", "not-returned@example.invalid", "Alice", + "fixture-org", self.name, int(time.time()) + 3600, + "fixture-access-secret", "fixture-refresh-secret") + self.tokens = {"fixture-alice": self.alice} + + def start_login(self, **kwargs): + raise NotImplementedError + + def complete_login(self, **kwargs): + raise NotImplementedError + + def verify_session(self, *, access_token): + return self.tokens.get(access_token) + + def refresh_session(self, **kwargs): + raise NotImplementedError + + def revoke_session(self, **kwargs): + raise NotImplementedError + + cls._provider = FixtureProvider() + register_provider(cls._provider) + app = FastAPI() + app.state.auth_required = True + app.middleware("http")(gated_auth_middleware) + app.include_router(native_api.router, prefix="/api/plugins/loopdy") + app.include_router(native_threads.router, prefix="/api/plugins/loopdy") + cls.client = TestClient(app) + + @classmethod + def tearDownClass(cls): + from hermes_cli.dashboard_auth.registry import unregister_global_provider + unregister_global_provider(cls._provider.name, cls._provider) + cls.client.close() + + def setUp(self): + from loopdy_plugin import native_threads + self.state = FakeState() + native_threads.register_state_resolver("default", self.state) + self.addCleanup(native_threads.register_state_resolver, "default", None) + temporary = tempfile.TemporaryDirectory(prefix="thread-mode-rest-") + self.addCleanup(temporary.cleanup) + self._env = patch.dict(os.environ, {"HERMES_HOME": temporary.name, "HOME": temporary.name}) + self._env.start() + self.addCleanup(self._env.stop) + + def _headers(self, token="fixture-alice"): + from loopdy_plugin import native_threads # noqa: F401 + context = self.client.get( + "/api/plugins/loopdy/native/context", + headers={"Authorization": "Bearer " + token}) + self.assertEqual(context.status_code, 200, context.text) + self.assertEqual(context.json()["servingProfileId"], "default") + return {"Authorization": "Bearer " + token, + "If-Match": context.headers["etag"], + "X-Loopdy-Request-ID": str(uuid.uuid4())} + + def test_flag_creates_and_updates_record(self): + response = self.client.post(self.PREFIX + "/flag", headers=self._headers(), + json={"session_id": "sess-1", "enabled": True}) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(response.json(), {"enabled": True, "ok": True, "session_id": "sess-1"}) + record = self.state.get(thread_mode.coordinator_key("sess-1")) + self.assertTrue(record["enabled"]) + response = self.client.post(self.PREFIX + "/flag", headers=self._headers(), + json={"session_id": "sess-1", "enabled": False}) + self.assertEqual(response.status_code, 200, response.text) + self.assertFalse(self.state.get(thread_mode.coordinator_key("sess-1"))["enabled"]) + + def test_register_adds_session_thread_and_index(self): + self.client.post(self.PREFIX + "/flag", headers=self._headers(), + json={"session_id": "sess-1", "enabled": True}) + response = self.client.post( + self.PREFIX + "/register", headers=self._headers(), + json={"coordinator_session_id": "sess-1", "name": "research", + "worker_session_id": "worker-1", "brief": "look things up"}) + self.assertEqual(response.status_code, 200, response.text) + thread = response.json()["thread"] + self.assertEqual(thread["name"], "research") + self.assertEqual(thread["kind"], "session") + self.assertEqual(thread["status"], "running") + self.assertEqual(thread["ids"]["worker_session_id"], "worker-1") + self.assertNotIn("handle", json.dumps(thread)) + index = self.state.get(thread_mode.worker_key("worker-1")) + self.assertEqual(index, {"coordinator_session_id": "sess-1", "thread": "research"}) + + def test_register_duplicate_name_is_409(self): + self.client.post(self.PREFIX + "/flag", headers=self._headers(), + json={"session_id": "sess-1", "enabled": True}) + body = {"coordinator_session_id": "sess-1", "name": "research", + "worker_session_id": "worker-1", "brief": "brief"} + self.assertEqual(self.client.post(self.PREFIX + "/register", headers=self._headers(), + json=body).status_code, 200) + duplicate = self.client.post(self.PREFIX + "/register", headers=self._headers(), + json={**body, "worker_session_id": "worker-2"}) + self.assertEqual(duplicate.status_code, 409, duplicate.text) + + def test_register_rejects_bad_name(self): + response = self.client.post( + self.PREFIX + "/register", headers=self._headers(), + json={"coordinator_session_id": "sess-1", "name": "Bad Name", + "worker_session_id": "worker-1", "brief": "brief"}) + self.assertEqual(response.status_code, 422, response.text) + + def test_roster_returns_record_without_handle(self): + self.client.post(self.PREFIX + "/flag", headers=self._headers(), + json={"session_id": "sess-1", "enabled": True}) + self.client.post(self.PREFIX + "/register", headers=self._headers(), + json={"coordinator_session_id": "sess-1", "name": "research", + "worker_session_id": "worker-1", "brief": "look things up"}) + response = self.client.get(self.PREFIX + "/roster", + params={"coordinator_session_id": "sess-1"}, + headers=self._headers()) + self.assertEqual(response.status_code, 200, response.text) + body = response.json() + self.assertEqual(body["coordinator_session_id"], "sess-1") + self.assertTrue(body["enabled"]) + self.assertEqual(len(body["threads"]), 1) + thread = body["threads"][0] + for field in ("name", "kind", "status", "brief", "ids", "notes", "result", "updated_at"): + self.assertIn(field, thread) + + def test_roster_unknown_session_is_404(self): + response = self.client.get(self.PREFIX + "/roster", + params={"coordinator_session_id": "missing"}, + headers=self._headers()) + self.assertEqual(response.status_code, 404, response.text) + + def test_requires_auth(self): + self.assertEqual(self.client.post(self.PREFIX + "/flag", + json={"session_id": "s", "enabled": True}).status_code, 401) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tools_registration.py b/tests/test_tools_registration.py index c33a61f..42ed917 100644 --- a/tests/test_tools_registration.py +++ b/tests/test_tools_registration.py @@ -121,6 +121,13 @@ def test_root_entrypoint_registers_model_schema_and_json_result(self): self.assertIn("cannot submit or publish", description) self.assertTrue(renderer_schema["parameters"]["properties"]["validateOnly"]) continue + if tool_name.startswith("thread_"): + # Thread-mode coordinator/worker tools (PROTOCOL.md): not renderers. + self.assertIn( + tool_name, ("thread_spawn", "thread_status", "thread_collect", "thread_note")) + self.assertFalse(renderer_schema["parameters"]["additionalProperties"]) + self.assertIn("thread", description.lower()) + continue self.assertIn("direct callable native Loopdy renderer", description) self.assertIn("visible in the current tool list", description) self.assertIn("tool_search", description) From 5e3232587276e9bd8d6dd10462aa91986df923b4 Mon Sep 17 00:00:00 2001 From: promptclickrun Date: Fri, 18 Sep 2026 11:34:54 -0500 Subject: [PATCH 2/2] fix(thread-mode): make chat drive orchestration Thread mode described manual worker creation as a normal path, which made the Threads surface feel like a setup form. Make the coordinator decide when parallel work helps, create and name workers itself, and return the assembled result in the ongoing chat. --- loopdy_plugin/thread_mode.py | 13 +++++++++---- tests/test_thread_mode.py | 11 +++++++++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/loopdy_plugin/thread_mode.py b/loopdy_plugin/thread_mode.py index c37c65d..c2b8008 100644 --- a/loopdy_plugin/thread_mode.py +++ b/loopdy_plugin/thread_mode.py @@ -451,10 +451,15 @@ def _coordinator_section(record: dict) -> str: "You are the coordinator of a Loopdy thread-mode session. " f"{len(threads)} worker thread(s) run in parallel.\n" f"{roster}\n" - "Delegate scoped work with `thread_spawn` while you have a live turn; the " - "Loopdy thread view can also launch session-backed workers out-of-turn. " - "Review progress with `thread_status` and gather finished work with " - "`thread_collect`; YOU assemble the final answer — never ask a worker to assemble it.\n" + "Treat this current coordinator chat as the user's single project conversation: " + "the user speaks naturally, and you automatically decide whether a request benefits " + "from parallel scoped worker threads. Answer simple requests directly. When parallel " + "work helps, choose thread names and briefs yourself and call `thread_spawn` during " + "the live turn, without asking the user to create or name threads. Monitor progress " + "with `thread_status`, gather finished work with `thread_collect`, and return the " + "assembled final answer in this current coordinator chat — never ask a worker to " + "assemble it. The Loopdy thread view's manual thread creation is an optional fallback " + "for session-backed workers, never the primary workflow.\n" "Memory discipline: ONLY you write durable profile memory (`MEMORY.md`). " "Workers never do; all cross-thread scratch lives in thread-mode plugin state " "and is injected into turns." diff --git a/tests/test_thread_mode.py b/tests/test_thread_mode.py index fbc88b0..97e4da3 100644 --- a/tests/test_thread_mode.py +++ b/tests/test_thread_mode.py @@ -424,6 +424,17 @@ def test_coordinator_turn_injects_roster_and_memory_discipline(self): self.assertIn("never ask a worker to assemble", context) self.assertIn("ONLY you write durable profile memory", context) + def test_coordinator_turn_makes_chat_the_primary_orchestration_surface(self): + record = thread_mode._ensure_coordinator(self.ctx.state, COORD) + self.ctx.state.set(thread_mode.coordinator_key(COORD), record) + + context = self._inject(session_id=COORD, parent_session_id="")["context"] + + self.assertIn("automatically decide", context) + self.assertIn("without asking the user to create or name threads", context) + self.assertIn("current coordinator chat", context) + self.assertIn("manual thread creation", context) + def test_non_thread_session_injects_nothing(self): self.assertIsNone(self._inject(session_id="plain", parent_session_id=""))