feat(kanban): stamp originating ACP session_id on tasks
Salvages #23208 by @awizemann. Tracks which chat session created a kanban task so clients can render a per-session board without falling back to tenant + time-window heuristics. - Schema: tasks gains nullable session_id TEXT column with index (additive migration in _migrate_add_optional_columns). - ACP: server.py exposes the originating session id via HERMES_SESSION_ID with save/restore around the agent loop. - Tool: kanban_create reads HERMES_SESSION_ID (with explicit override). - CLI: 'hermes kanban list --session <id>' filter; JSON output exposes session_id.
This commit is contained in:
@@ -1090,6 +1090,80 @@ class TestPrompt:
|
||||
]
|
||||
assert any(update.session_update == "agent_message_chunk" for update in updates)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_propagates_hermes_session_id_env(self, agent, monkeypatch):
|
||||
"""ACP must propagate the originating session id to the agent loop
|
||||
via ``HERMES_SESSION_ID`` so tools that want to stamp side-effects
|
||||
with it (e.g. ``kanban_create``) can read the env var inside
|
||||
``run_conversation``. The variable must be visible during the
|
||||
agent call AND restored afterwards so a re-used executor thread
|
||||
doesn't leak one session's id into another."""
|
||||
# Pre-condition: env is clean.
|
||||
monkeypatch.delenv("HERMES_SESSION_ID", raising=False)
|
||||
|
||||
new_resp = await agent.new_session(cwd=".")
|
||||
state = agent.session_manager.get_session(new_resp.session_id)
|
||||
|
||||
captured: dict[str, str | None] = {}
|
||||
|
||||
def mock_run(user_message, conversation_history=None, task_id=None, **kwargs):
|
||||
# Inside the agent loop the env var must reflect the active
|
||||
# ACP session id. ``task_id`` is also the session id at this
|
||||
# boundary; assert both for symmetry.
|
||||
captured["env"] = os.environ.get("HERMES_SESSION_ID")
|
||||
captured["task_id"] = task_id
|
||||
return {"final_response": "ok", "messages": []}
|
||||
|
||||
state.agent.run_conversation = mock_run
|
||||
|
||||
mock_conn = MagicMock(spec=acp.Client)
|
||||
mock_conn.session_update = AsyncMock()
|
||||
agent._conn = mock_conn
|
||||
|
||||
prompt = [TextContentBlock(type="text", text="hi")]
|
||||
await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
|
||||
|
||||
assert captured["env"] == new_resp.session_id, (
|
||||
"HERMES_SESSION_ID must be set to the originating ACP session id "
|
||||
"while the agent loop is running"
|
||||
)
|
||||
assert captured["task_id"] == new_resp.session_id
|
||||
# Post-condition: must be restored to the prior value (None here).
|
||||
assert os.environ.get("HERMES_SESSION_ID") is None, (
|
||||
"HERMES_SESSION_ID must be restored after the agent call so "
|
||||
"a re-used executor thread doesn't leak the id into the next "
|
||||
"session's tools"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_restores_prior_hermes_session_id(self, agent, monkeypatch):
|
||||
"""If the env already had HERMES_SESSION_ID set (e.g. nested
|
||||
agent loops), the prior value must be restored after the inner
|
||||
prompt completes — not popped, not left at the inner id."""
|
||||
monkeypatch.setenv("HERMES_SESSION_ID", "outer-sess")
|
||||
|
||||
new_resp = await agent.new_session(cwd=".")
|
||||
state = agent.session_manager.get_session(new_resp.session_id)
|
||||
|
||||
captured: dict[str, str | None] = {}
|
||||
|
||||
def mock_run(*args, **kwargs):
|
||||
captured["inner"] = os.environ.get("HERMES_SESSION_ID")
|
||||
return {"final_response": "ok", "messages": []}
|
||||
|
||||
state.agent.run_conversation = mock_run
|
||||
|
||||
mock_conn = MagicMock(spec=acp.Client)
|
||||
mock_conn.session_update = AsyncMock()
|
||||
agent._conn = mock_conn
|
||||
|
||||
prompt = [TextContentBlock(type="text", text="hi")]
|
||||
await agent.prompt(prompt=prompt, session_id=new_resp.session_id)
|
||||
|
||||
assert captured["inner"] == new_resp.session_id
|
||||
# Outer scope must be restored.
|
||||
assert os.environ.get("HERMES_SESSION_ID") == "outer-sess"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_prompt_does_not_duplicate_streamed_final_message(self, agent):
|
||||
"""If ACP already streamed response chunks, final_response should not be sent again."""
|
||||
|
||||
@@ -156,6 +156,48 @@ def test_run_slash_tenant_filter(kanban_home):
|
||||
assert "biz-b task" in b and "biz-a task" not in b
|
||||
|
||||
|
||||
def test_run_slash_session_filter(kanban_home):
|
||||
"""`hermes kanban list --session <id>` filters by the originating
|
||||
chat session id stamped on tasks created from inside an ACP loop."""
|
||||
from hermes_cli import kanban_db as kb
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(
|
||||
conn, title="from sess-1 a", assignee="alice", session_id="sess-1"
|
||||
)
|
||||
kb.create_task(
|
||||
conn, title="from sess-1 b", assignee="alice", session_id="sess-1"
|
||||
)
|
||||
kb.create_task(
|
||||
conn, title="from sess-2", assignee="alice", session_id="sess-2"
|
||||
)
|
||||
kb.create_task(conn, title="cli only", assignee="alice")
|
||||
out_1 = kc.run_slash("list --session sess-1")
|
||||
out_2 = kc.run_slash("list --session sess-2")
|
||||
assert "from sess-1 a" in out_1
|
||||
assert "from sess-1 b" in out_1
|
||||
assert "from sess-2" not in out_1
|
||||
assert "cli only" not in out_1
|
||||
assert "from sess-2" in out_2
|
||||
assert "from sess-1 a" not in out_2
|
||||
|
||||
|
||||
def test_kanban_list_json_includes_session_id(kanban_home):
|
||||
"""JSON output exposes `session_id` so external clients (Scarf, web
|
||||
dashboards) don't need a side query to filter by chat session."""
|
||||
from hermes_cli import kanban_db as kb
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(
|
||||
conn, title="acp task", assignee="alice", session_id="acp-x"
|
||||
)
|
||||
raw = kc.run_slash("list --json")
|
||||
payload = json.loads(raw)
|
||||
assert any(
|
||||
row.get("title") == "acp task"
|
||||
and row.get("session_id") == "acp-x"
|
||||
for row in payload
|
||||
)
|
||||
|
||||
|
||||
def test_run_slash_usage_error_returns_message(kanban_home):
|
||||
# Missing required argument for create
|
||||
out = kc.run_slash("create")
|
||||
|
||||
@@ -1035,6 +1035,76 @@ def test_tenant_propagates_to_events(kanban_home):
|
||||
assert created and created[0].payload.get("tenant") == "biz-a"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Originating session id (ACP propagation)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_create_task_stamps_session_id(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(
|
||||
conn, title="from chat", session_id="acp-sess-123"
|
||||
)
|
||||
t = kb.get_task(conn, tid)
|
||||
assert t is not None
|
||||
assert t.session_id == "acp-sess-123"
|
||||
|
||||
|
||||
def test_create_task_session_id_defaults_to_none(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
tid = kb.create_task(conn, title="cli-created")
|
||||
t = kb.get_task(conn, tid)
|
||||
assert t is not None
|
||||
assert t.session_id is None
|
||||
|
||||
|
||||
def test_session_id_filters_listings(kanban_home):
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(conn, title="s1-a", session_id="sess-1")
|
||||
kb.create_task(conn, title="s1-b", session_id="sess-1")
|
||||
kb.create_task(conn, title="s2-a", session_id="sess-2")
|
||||
kb.create_task(conn, title="cli-only") # no session
|
||||
sess1 = kb.list_tasks(conn, session_id="sess-1")
|
||||
sess2 = kb.list_tasks(conn, session_id="sess-2")
|
||||
unscoped = kb.list_tasks(conn)
|
||||
assert sorted(t.title for t in sess1) == ["s1-a", "s1-b"]
|
||||
assert [t.title for t in sess2] == ["s2-a"]
|
||||
# Unscoped list still returns everything (legacy NULL rows visible).
|
||||
assert len(unscoped) == 4
|
||||
|
||||
|
||||
def test_session_id_index_exists(kanban_home):
|
||||
"""The migration creates an index on session_id for cheap per-session
|
||||
list queries on busy boards. Without it, a chat-scoped poll would
|
||||
full-scan the tasks table."""
|
||||
with kb.connect() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='index' "
|
||||
"AND tbl_name='tasks'"
|
||||
).fetchall()
|
||||
names = {r["name"] for r in rows}
|
||||
assert "idx_tasks_session_id" in names
|
||||
|
||||
|
||||
def test_session_id_compose_with_tenant_filter(kanban_home):
|
||||
"""A client may want both `tenant=scarf:foo` AND `session=acp-x` —
|
||||
the filters must AND, not replace."""
|
||||
with kb.connect() as conn:
|
||||
kb.create_task(
|
||||
conn, title="match", tenant="scarf:foo", session_id="acp-x"
|
||||
)
|
||||
kb.create_task(
|
||||
conn, title="wrong-tenant", tenant="other", session_id="acp-x"
|
||||
)
|
||||
kb.create_task(
|
||||
conn, title="wrong-session",
|
||||
tenant="scarf:foo", session_id="acp-y",
|
||||
)
|
||||
rows = kb.list_tasks(
|
||||
conn, tenant="scarf:foo", session_id="acp-x"
|
||||
)
|
||||
assert [t.title for t in rows] == ["match"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared-board path resolution (issue #19348)
|
||||
#
|
||||
@@ -1557,7 +1627,8 @@ def test_migrate_add_optional_columns_tolerates_concurrent_migration(kanban_home
|
||||
workflow_template_id TEXT,
|
||||
current_step_key TEXT,
|
||||
skills TEXT,
|
||||
max_retries INTEGER
|
||||
max_retries INTEGER,
|
||||
session_id TEXT
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
@@ -768,6 +768,75 @@ def test_create_happy_path(worker_env):
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_create_stamps_session_id_from_env(monkeypatch, worker_env):
|
||||
"""When the agent loop runs under ACP, the server propagates the
|
||||
originating chat session id via HERMES_SESSION_ID. ``kanban_create``
|
||||
reads it and stamps the new task so clients can render a per-session
|
||||
board (issue: ACP session linkage on kanban tasks)."""
|
||||
monkeypatch.setenv("HERMES_SESSION_ID", "acp-sess-abc")
|
||||
from tools import kanban_tools as kt
|
||||
from hermes_cli import kanban_db as kb
|
||||
out = kt._handle_create({
|
||||
"title": "from chat",
|
||||
"assignee": "peer",
|
||||
"parents": [worker_env],
|
||||
})
|
||||
d = json.loads(out)
|
||||
assert d["ok"] is True
|
||||
conn = kb.connect()
|
||||
try:
|
||||
new_task = kb.get_task(conn, d["task_id"])
|
||||
assert new_task.session_id == "acp-sess-abc"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_create_session_id_arg_overrides_env(monkeypatch, worker_env):
|
||||
"""An explicit ``session_id`` arg from the model wins over the env
|
||||
propagation. Edge case but exercised: a tool call could carry a
|
||||
different session id (e.g. cross-session linking) and the explicit
|
||||
arg should not be silently overwritten."""
|
||||
monkeypatch.setenv("HERMES_SESSION_ID", "from-env")
|
||||
from tools import kanban_tools as kt
|
||||
from hermes_cli import kanban_db as kb
|
||||
out = kt._handle_create({
|
||||
"title": "explicit override",
|
||||
"assignee": "peer",
|
||||
"parents": [worker_env],
|
||||
"session_id": "explicit-arg",
|
||||
})
|
||||
d = json.loads(out)
|
||||
assert d["ok"] is True
|
||||
conn = kb.connect()
|
||||
try:
|
||||
new_task = kb.get_task(conn, d["task_id"])
|
||||
assert new_task.session_id == "explicit-arg"
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_create_session_id_absent_when_env_unset(monkeypatch, worker_env):
|
||||
"""No env var, no arg → session_id stays NULL. Important for backwards
|
||||
compatibility: pre-ACP-propagation hosts and CLI-driven creates must
|
||||
not accidentally inherit a stale id."""
|
||||
monkeypatch.delenv("HERMES_SESSION_ID", raising=False)
|
||||
from tools import kanban_tools as kt
|
||||
from hermes_cli import kanban_db as kb
|
||||
out = kt._handle_create({
|
||||
"title": "no session",
|
||||
"assignee": "peer",
|
||||
"parents": [worker_env],
|
||||
})
|
||||
d = json.loads(out)
|
||||
assert d["ok"] is True
|
||||
conn = kb.connect()
|
||||
try:
|
||||
new_task = kb.get_task(conn, d["task_id"])
|
||||
assert new_task.session_id is None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_create_rejects_no_title(worker_env):
|
||||
from tools import kanban_tools as kt
|
||||
assert json.loads(kt._handle_create({"assignee": "x"})).get("error")
|
||||
|
||||
Reference in New Issue
Block a user