feat(b1): fix find_workspace_uuid tier-3 identity lookup and add test_b1_tier3_identity.py (100% PASS)

This commit is contained in:
2026-08-05 23:55:21 +09:00
parent b6c41e6486
commit 924e77e8b8
5 changed files with 903 additions and 29 deletions
+25 -20
View File
@@ -1102,7 +1102,8 @@ verify_tui_viewport() {
# (claude_session_id_own / agy_conversation_id_own)
# 2) on-disk scan scoped to this workspace
# (claude: ~/.claude/projects/<key>/*.jsonl ; agy: last_conversations.json[cwd])
# 3) agent_identities cache, ONLY when its project_cwd == this workspace
# 3) agent_identities cache in d (primary) or $YAML_PATH (fallback), ONLY when its project_cwd == this workspace.
# (Note: DB is authority, YAML is mirror; tier-3 never prioritizes mirror over DB)
# Prints the UUID on stdout (empty line if none). Always exits 0.
# ---------------------------------------------------------------------------
find_workspace_uuid() {
@@ -1263,23 +1264,27 @@ elif agent == 'cline':
if cand and verify_session_uuid(ws, agent, cand):
emit(cand)
ai = {}
db_path = f"{mam_dir}/agent-sessions.db" if 'mam_dir' in locals() else os.path.join(ws, ".mam", "agent-sessions.db")
yaml_path = f"{mam_dir}/agent-sessions.yaml" if 'mam_dir' in locals() else os.path.join(ws, ".mam", "agent-sessions.yaml")
try:
import yaml
if os.path.exists(db_path):
conn = sqlite3.connect(db_path, timeout=60.0)
row = conn.execute('SELECT data FROM state WHERE id=1').fetchone()
if row:
ai = json.loads(row[0]).get('agent_identities', {})
conn.close()
elif os.path.exists(yaml_path):
with open(yaml_path) as f:
d = yaml.safe_load(f) or {}
ai = d.get('agent_identities', {})
except Exception as e:
print(f"WARN: tier-3 identity lookup failed: {e}", file=sys.stderr)
ai = d.get('agent_identities') if isinstance(d, dict) else None
if not isinstance(ai, dict) or not ai:
ai = {}
try:
yaml_path = os.environ['YAML_PATH']
db_path = os.path.splitext(yaml_path)[0] + '.db'
if os.path.exists(db_path):
conn = sqlite3.connect(db_path, timeout=60.0)
row = conn.execute('SELECT data FROM state WHERE id=1').fetchone()
if row:
ai = json.loads(row[0]).get('agent_identities') or {}
conn.close()
elif os.path.exists(yaml_path):
import yaml
with open(yaml_path) as f:
_ydoc = yaml.safe_load(f) or {}
ai = _ydoc.get('agent_identities') or {}
except Exception as e:
print(f"WARN: tier-3 identity lookup failed: {e}", file=sys.stderr)
if not isinstance(ai, dict):
ai = {}
ai_agent = ai.get(agent) or {}
if ai_agent.get('project_cwd') == ws:
@@ -1292,11 +1297,11 @@ if ai_agent.get('project_cwd') == ws:
if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"):
emit(cand)
elif agent == 'hermes':
cand = ai_agent.get('session_id') or ai.get('conversation_id')
cand = ai_agent.get('session_id') or ai_agent.get('conversation_id')
if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"):
emit(cand)
elif agent == 'cline':
cand = ai_agent.get('session_id') or ai.get('conversation_id')
cand = ai_agent.get('session_id') or ai_agent.get('conversation_id')
if cand and verify_session_uuid(ws, agent, cand, mode="revalidate"):
emit(cand)