fix(mobile): wire plugin catalog and session-scoped subagent controls

tests/tui_gateway/test_mobile_subagents.py

@@ -0,0 +1,76 @@
+from types import SimpleNamespace
+
+from tui_gateway import mobile_subagents as mobile
+
+
+def test_roster_is_session_and_transport_scoped(monkeypatch):
+    from tools import delegate_tool as delegate
+    transport, session = object(), {}
+    own = dict(subagent_id="own", owner_session_id="s", owner_transport=transport,
+               owner_session_record=session, agent=object(), goal="test", started_at=1)
+    other = {**own, "subagent_id": "other", "owner_transport": object()}
+    reused = {**own, "subagent_id": "reused", "owner_session_record": {}}
+    monkeypatch.setattr(delegate, "_active_subagents", {"own": own, "other": other, "reused": reused})
+    rows = mobile.owned_records("s", transport, session)
+    assert [row["subagent_id"] for row in rows] == ["own"]
+    assert mobile.owned_records("s", None, session) == []
+    assert mobile.owned_records("wrong", transport, session) == []
+    public = mobile.roster(rows)["subagents"][0]
+    assert public["elapsed_seconds"] > 0
+    assert "agent" not in public and "owner_session_record" not in public
+
+
+def test_tail_is_bounded_and_unknown_child_is_unavailable(tmp_path, monkeypatch):
+    from tools import delegation_live_log
+    monkeypatch.setattr(delegation_live_log, "live_transcript_root", lambda: tmp_path)
+    path = tmp_path / "transcript.jsonl"
+    path.write_bytes(b"x" * 20000 + b"end")
+    rows = [dict(subagent_id="own", agent=SimpleNamespace(_live_transcript_path=str(path)))]
+    result = mobile.transcript_tail(rows, "own")
+    assert result["available"] and result["truncated"]
+    assert result["bytes_read"] == 16384 and result["text"].endswith("end")
+    assert not mobile.transcript_tail(rows, "other")["available"]
+    path.unlink()
+    assert not mobile.transcript_tail(rows, "own")["available"]
+
+
+def test_tail_rejects_outside_log_root(tmp_path, monkeypatch):
+    from tools import delegation_live_log
+    monkeypatch.setattr(delegation_live_log, "live_transcript_root", lambda: tmp_path / "logs")
+    secret = tmp_path / "private.txt"
+    secret.write_text("not a transcript")
+    rows = [dict(subagent_id="own", agent=SimpleNamespace(_live_transcript_path=str(secret)))]
+    assert not mobile.transcript_tail(rows, "own")["available"]
+
+
+def test_roster_survives_resume_but_tail_keeps_exact_authority(monkeypatch):
+    from tools import delegate_tool as delegate
+    parent = SimpleNamespace(session_id="conversation")
+    record = dict(subagent_id="child", owner_session_id="old-runtime",
+                  owner_agent_session_id="conversation", agent=None)
+    monkeypatch.setattr(delegate, "_active_subagents", {"child": record})
+    session = {"agent": parent}
+    assert mobile.owned_records("new-runtime", object(), session) == []
+    assert len(mobile.owned_records("new-runtime", object(), session, include_lineage=True)) == 1
+
+
+def test_registered_rpcs_reject_wrong_transport(monkeypatch):
+    from tui_gateway import server
+    monkeypatch.setattr(server, "_sess_nowait", lambda params, rid: ({"session_key": "s"}, None))
+    monkeypatch.setattr(server, "_current_session_steer_authority", lambda sid: (None, None))
+    for name in ("subagent.list", "subagent.tail", "subagent.interrupt"):
+        response = server._methods[name](1, {"session_id": "s", "subagent_id": "child"})
+        assert response["error"]["code"] == 4001
+
+
+def test_registered_list_returns_owned_children(monkeypatch):
+    from tui_gateway import server
+    from tools import delegate_tool as delegate
+    transport, session = object(), {"session_key": "s"}
+    record = dict(subagent_id="child", owner_session_id="s", owner_transport=transport,
+                  owner_session_record=session, goal="test")
+    monkeypatch.setattr(delegate, "_active_subagents", {"child": record})
+    monkeypatch.setattr(server, "_sess_nowait", lambda params, rid: (session, None))
+    monkeypatch.setattr(server, "_current_session_steer_authority", lambda sid: (transport, session))
+    response = server._methods["subagent.list"](1, {"session_id": "s"})
+    assert response["result"]["subagents"][0]["subagent_id"] == "child"