100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
# workspace_uuid.py — workspace UUID discovery logic
|
|
# Extracted from lib.sh find_workspace_uuid PYEOF block
|
|
|
|
import os, sys, json
|
|
from lib_py.verify_session import verify_session_uuid, mam_orchestrator_uuids, mam_row_own_uuid
|
|
|
|
OWN_KEY = {
|
|
'claude': 'claude_session_id_own',
|
|
'agy': 'agy_conversation_id_own',
|
|
'hermes': 'hermes_conversation_id_own',
|
|
'grok': 'grok_session_id_own',
|
|
'opencode': 'opencode_session_id_own',
|
|
'codex': 'codex_session_id_own'
|
|
}
|
|
|
|
from lib_py.paths import resolve_home
|
|
|
|
def find_workspace_uuid(ws, agent, target='', state=None, home=None, claude_dir=None, exclude_ids=()):
|
|
d = state if isinstance(state, dict) else {}
|
|
home = resolve_home(home)
|
|
claude_project_dir = claude_dir or os.environ.get('CLAUDE_PROJECT_DIR', f"{home}/.claude/projects")
|
|
|
|
running_ids = set()
|
|
for s_item in d.get('herdr_sessions', []):
|
|
if s_item.get('status') == 'running':
|
|
if target and s_item.get('name') == target:
|
|
continue
|
|
for k in ['claude_session_id_own', 'agy_conversation_id_own', 'hermes_conversation_id_own', 'grok_session_id_own', 'opencode_session_id_own', 'codex_session_id_own']:
|
|
val = s_item.get(k)
|
|
if val:
|
|
running_ids.add(val)
|
|
|
|
orchestrator_ids = set(mam_orchestrator_uuids())
|
|
if "MAM_ORCHESTRATOR_UUIDS" not in os.environ:
|
|
raw_orc = d.get('orchestrator_uuids')
|
|
if isinstance(raw_orc, list) and all(isinstance(x, str) and x.strip() for x in raw_orc):
|
|
orchestrator_ids |= set(raw_orc)
|
|
if target:
|
|
for s_item in d.get('herdr_sessions', []):
|
|
if s_item.get('name') == target:
|
|
own = mam_row_own_uuid(s_item)
|
|
if own:
|
|
orchestrator_ids.discard(own)
|
|
|
|
excluded = running_ids | orchestrator_ids | {u for u in (exclude_ids or ()) if u}
|
|
|
|
# Fetch all sessions matching this workspace
|
|
sessions = []
|
|
for s in d.get('herdr_sessions', []):
|
|
s_cwd = (s.get('pane') or {}).get('cwd', '') if isinstance(s, dict) else ''
|
|
if s_cwd and (s_cwd == ws or os.path.realpath(s_cwd) == os.path.realpath(ws)):
|
|
sessions.append(s)
|
|
|
|
if target:
|
|
for s in sessions:
|
|
if s.get('name') != target:
|
|
continue
|
|
cand = s.get(OWN_KEY.get(agent, ''), None)
|
|
if cand and verify_session_uuid(ws, agent, cand, s, home_dir=home, claude_dir=claude_project_dir, mode="revalidate"):
|
|
if cand not in excluded:
|
|
return cand
|
|
|
|
for s in sessions:
|
|
name = s.get('name', '')
|
|
if name.endswith(f"-creator-{agent}"):
|
|
cand = s.get(OWN_KEY.get(agent, ''))
|
|
if cand and verify_session_uuid(ws, agent, cand, s, home_dir=home, claude_dir=claude_project_dir, mode="revalidate"):
|
|
if cand not in excluded:
|
|
return cand
|
|
|
|
from lib_py.agents.registry import get_adapter
|
|
from lib_py.agents.base import DiscoveryContext
|
|
|
|
adapter = get_adapter(agent)
|
|
if adapter:
|
|
ctx = DiscoveryContext(workspace=ws, agent_name=agent, home_dir=home, claude_dir=claude_project_dir)
|
|
for cand in adapter.discover(ctx):
|
|
if cand and cand not in excluded:
|
|
return cand
|
|
|
|
return ''
|
|
|
|
|
|
def find_workspace_uuid_main():
|
|
ws = os.environ['WS_ABS']
|
|
agent = os.environ['AGENT']
|
|
home = resolve_home()
|
|
claude_project_dir = os.environ.get('CLAUDE_PROJECT_DIR', f"{home}/.claude/projects")
|
|
target = os.environ.get('TARGET_SESSION', '')
|
|
|
|
try:
|
|
d = json.loads(os.environ.get('MAM_STATE_JSON', '{}'))
|
|
except Exception:
|
|
d = {}
|
|
|
|
print(find_workspace_uuid(ws, agent, target, d, home, claude_project_dir))
|
|
|
|
if __name__ == '__main__':
|
|
find_workspace_uuid_main()
|