diff --git a/harbor/hooks.py b/harbor/hooks.py index e2bccf3..ef6ad2f 100644 --- a/harbor/hooks.py +++ b/harbor/hooks.py @@ -63,7 +63,7 @@ ALLOWED_COMMANDS = {"brief", "inbox", "ack", "thread", "threads", "search", "tai "tasks", "pins", "decisions", "handoffs", "handoff", "item", "pending-acks", "wait", "doing", "send", "reply", "answer", "questions", "context", "notes", "note", "digest", "attention", "stop", "who", "feed", "summarize", "git", "templates", "pulse", "metrics", "wins"} -ALLOWED_TOOLS = {"brief", "inbox", "ack", "thread", "threads", "search", "status", "queue", "claims", "tasks", "handoff", +ALLOWED_TOOLS = {"brief", "i_am", "mark_read", "inbox", "ack", "thread", "threads", "search", "status", "queue", "claims", "tasks", "handoff", "send", "reply", "wait_for_messages", "doing", "questions", "answer", "context", "notes", "note", "digest", "pin", "close_thread", "stop", "summarize", "who_knows", "git_status"} TOOL_NAME = re.compile(r"^(?:mcp__harbor__|harbor[_.])([a-z_]+)$") @@ -276,6 +276,10 @@ def _run(event, harbor, agent, host, payload): try: # bookkeeping is best effort from .launch import describe_target tool_input = payload.get("tool_input") or payload.get("input") or {} + harbor_tool = _harbor_tool(str(tool)) + if harbor_tool is not None: # tell the app's shared Harbor server which session is calling + from . import sessions + sessions.leave(harbor, host, harbor_tool, tool_input if isinstance(tool_input, dict) else {}, agent) paths = harbor.edited_paths(tool, tool_input) harbor.activity(agent, tool, host, target=describe_target(tool, tool_input, paths)) if not stops: diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index b75a59b..139af26 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -5,6 +5,7 @@ Run with the agent's identity fixed in its environment (it cannot be changed fro The project comes from $HARBOR_PROJECT or the .harbor.toml above the working directory. The human is refused. """ import asyncio +import contextvars import os import threading import time @@ -14,8 +15,9 @@ os.environ.setdefault("FASTMCP_CHECK_FOR_UPDATES", "off") # no network call os.environ.setdefault("FASTMCP_SHOW_SERVER_BANNER", "false") from fastmcp import FastMCP # noqa: E402 from fastmcp.exceptions import ToolError # noqa: E402 +from fastmcp.server.middleware import Middleware # noqa: E402 -from . import projects # noqa: E402 +from . import projects, sessions # noqa: E402 from .core import OPEN_ITEM_STATES, Harbor, HarborError # noqa: E402 INSTRUCTIONS = """Harbor is the team's shared coordination store: messages in threads, acknowledgements, claims, a @@ -55,6 +57,30 @@ ClaimKind = Literal["draft", "naming", "edit", "review", "band"] mcp = FastMCP("harbor", instructions=INSTRUCTIONS) _lock = threading.Lock() _state = {"seen": 0.0} +_caller = contextvars.ContextVar("harbor_caller", default=None) # the session behind the current tool call + + +class SessionIdentity(Middleware): + """One app may share this server between several sessions (Codex and Codex2). The app's PreToolUse hook names + the calling session in a ticket; take it so this call runs as that agent (harbor.sessions).""" + + async def on_call_tool(self, context, call_next): + agent = None + try: + message = context.message + with _lock: + agent = sessions.take(harbor(), os.environ.get("HARBOR_HOST", ""), message.name, + dict(message.arguments or {})) + except Exception: # noqa: BLE001 - identity falls back to the server's own + agent = None + token = _caller.set(agent) + try: + return await call_next(context) + finally: + _caller.reset(token) + + +mcp.add_middleware(SessionIdentity()) def harbor(): @@ -67,9 +93,9 @@ def harbor(): def me(): h = harbor() - # a session bound with i_am, else the agent its working folder names (when it runs in the same app as the - # installed one: Codex and Codex2 share the Codex app), else the installed one - agent = _state.get("bound") or h.project.session_identity(os.environ.get("HARBOR_AGENT"), os.getcwd()) or "" + # the session its app's hook named for this call (several sessions can share this server), else the agent the + # server's working folder names when it runs in the same app as the installed one, else the installed one + agent = (_caller.get() or h.project.session_identity(os.environ.get("HARBOR_AGENT"), os.getcwd()) or "") if agent not in h.project.ai_names: raise HarborError(f"HARBOR_AGENT must be one of {', '.join(h.project.ai_names)} (set it in this server's MCP " "config)") @@ -125,22 +151,19 @@ def run(fn): @mcp.tool def i_am(agent: str) -> dict: - """Say which of the project's agents this session is, when one app runs several (Codex and Codex2). Call it once - at the start of the session if your instructions or your working folder name you; only an agent of the same - app is accepted, and a server already bound to another agent refuses.""" + """Check which of the project's agents Harbor takes this session to be, when one app runs several (Codex and + Codex2). Call it at the start of the session with the name your instructions give you. Harbor knows the session + from its working folder (through the app's hooks); if the answer is not you, stop and tell the user.""" def do(h): - pinned = os.environ.get("HARBOR_AGENT") or "" if agent not in h.project.ai_names: raise HarborError(f"{agent} is not one of {', '.join(h.project.ai_names)}") - if pinned and h.project.host_of(agent) != h.project.host_of(pinned): - raise HarborError(f"{agent} runs in another app ({h.project.host_of(agent) or 'unknown'}), not this one") - bound = _state.get("bound") - if bound and bound != agent: - raise HarborError(f"this Harbor server is already {bound}: the app shares one server between sessions, " - "so give each session its own app profile instead") - _state["bound"] = agent - return {"you_are": agent, "working_folder": os.getcwd(), - "folder_says": h.project.identity_for(os.getcwd())} + seen = me() + out = {"you_are": seen, "matches": seen == agent} + if seen != agent: + out["fix"] = (f"Harbor sees this session as {seen}, not {agent}. Check that this session works in " + f"{agent}'s own worktree and that the app's Harbor hooks are installed and trusted; do no " + "work until it matches.") + return out return run(do) diff --git a/harbor/sessions.py b/harbor/sessions.py new file mode 100644 index 0000000..54a3e66 --- /dev/null +++ b/harbor/sessions.py @@ -0,0 +1,51 @@ +"""Which agent made this MCP call, when one app runs several sessions (Codex and Codex2) on one shared Harbor server. + +An app starts one Harbor MCP server and shares it between its sessions, so the server cannot tell them apart. The +app's PreToolUse hook can: it runs before every tool call with the session's working folder. So the hook leaves a +ticket (app, tool, arguments, agent) and the server takes the matching ticket when the call arrives - oldest first, +within two minutes. Without a ticket (hooks not installed or not trusted) the server falls back to its own identity.""" +import json +import time + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS session_tickets(id INTEGER PRIMARY KEY, host TEXT NOT NULL, tool TEXT NOT NULL, + args TEXT NOT NULL, agent TEXT NOT NULL, at REAL NOT NULL, used INTEGER NOT NULL DEFAULT 0); +CREATE INDEX IF NOT EXISTS session_tickets_open ON session_tickets(host, tool, used, at); +""" +MAX_AGE = 120.0 + + +def args_key(args): + return json.dumps(args or {}, sort_keys=True, separators=(",", ":"), default=str) + + +def leave(harbor, host, tool, args, agent, now=None): + now = time.time() if now is None else now + + def do(c): + c.execute("DELETE FROM session_tickets WHERE at < ?", (now - 10 * MAX_AGE,)) + c.execute("INSERT INTO session_tickets(host, tool, args, agent, at) VALUES (?,?,?,?,?)", + (host or "", tool, args_key(args), agent, now)) + ensure(harbor) + harbor._write(do) + + +def take(harbor, host, tool, args, now=None): + """The agent whose hook announced this exact call, or None.""" + now = time.time() if now is None else now + ensure(harbor) + + def do(c): + row = c.execute("SELECT id, agent FROM session_tickets WHERE host=? AND tool=? AND args=? AND used=0 AND at>=? " + "ORDER BY at, id LIMIT 1", (host or "", tool, args_key(args), now - MAX_AGE)).fetchone() + if not row: + return None + c.execute("UPDATE session_tickets SET used=1 WHERE id=?", (row[0],)) + return row[1] + return harbor._write(do) + + +def ensure(harbor): + if not getattr(harbor, "_session_tickets_ready", False): + harbor.conn.executescript(SCHEMA) + harbor._session_tickets_ready = True diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py index be49825..250cc30 100644 --- a/tests/test_interfaces.py +++ b/tests/test_interfaces.py @@ -288,6 +288,41 @@ class ServerWork(unittest.TestCase): class Mcp(unittest.TestCase): + def test_shared_server_takes_the_calling_session_from_the_hook_ticket(self): + """One app's server shared by two sessions: each call runs as the agent its hook named in a ticket.""" + from fastmcp import Client + from harbor import mcp_server, sessions + d = temp_dir() + project = make_home(d)["test"] + mcp_server._state.clear() + mcp_server._state["project"] = project + saved = {k: os.environ.get(k) for k in ("HARBOR_AGENT", "HARBOR_HOST")} + os.environ.update(HARBOR_AGENT="Codex", HARBOR_HOST="codex") + + async def scenario(): + async with Client(mcp_server.mcp) as client: + sessions.leave(mcp_server.harbor(), "codex", "doing", {"text": "second session"}, "DeepSeek") + await client.call_tool("doing", {"text": "second session"}) # the ticket names DeepSeek + await client.call_tool("doing", {"text": "first session"}) # no ticket: the server's own + status = (await client.call_tool("status", {})).data + doing = {a["name"]: a["doing"] for a in status["agents"]} + self.assertEqual(doing["DeepSeek"], "second session") + self.assertEqual(doing["Codex"], "first session") + sessions.leave(mcp_server.harbor(), "codex", "i_am", {"agent": "DeepSeek"}, "DeepSeek") + self.assertTrue((await client.call_tool("i_am", {"agent": "DeepSeek"})).data["matches"]) + self.assertFalse((await client.call_tool("i_am", {"agent": "DeepSeek"})).data["matches"]) # used up + + try: + asyncio.run(scenario()) + finally: + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + mcp_server._state.pop("h").close() + mcp_server._state.clear() + def test_tools_bind_identity_and_round_trip(self): from fastmcp import Client from fastmcp.exceptions import ToolError