Shared app servers: the PreToolUse hook leaves a ticket naming the calling session, the MCP server takes it per call (middleware), so two sessions of one app on one Harbor server act as themselves; i_am checks instead of binding and is allowed under a stop

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