feat(a4,c3b): complete A-4 Phase 2 agent knowledge migration & Option B isolation removal

- Option B / C-3b: Deprecate isolation.root and remove its 4 legacy consumers in lib.sh, workspace_uuid.py, verify_session.py, and stop_session.sh.
- M2~M7 Migration: Implement artifact_path, verify_artifact, purge_artifacts, spawn_spec, resume_spec, auth_ok, discover, ready_tokens, exit_key, and delegate_agent_key across BaseAgentAdapter and all 4 adapters (Claude, Agy, Hermes, Cline).
- Migrate shell duplications in lib.sh, create_session.sh, resume_session.sh, stop_session.sh, and reconcile.sh to python -m lib_py.agents facts.
- Hardening & Corrections: Fix Cline resume_spec flag to '-i --id', apply shlex.quote across all facts variables with MAM_ prefix, add safe fallback in wait_for_tui_ready.
- Add comprehensive contract tests in tests/test_a4_adapter_contract.py and sync IMPROVEMENTS.md / LOG.md.
- Verified: 100% PASS across unit, component, contract, and integration tests.
This commit is contained in:
2026-08-16 23:15:24 +09:00
parent 971f14ad3f
commit b4821fafa8
21 changed files with 904 additions and 531 deletions
+100 -3
View File
@@ -1,6 +1,7 @@
# claude.py — Claude Code agent adapter
from lib_py.agents.base import BaseAgentAdapter
import os, json, glob, subprocess
from typing import Optional, Any
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
from lib_py.verify_session import workspace_key
class ClaudeAgentAdapter(BaseAgentAdapter):
@property
@@ -11,6 +12,22 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
def own_key(self) -> str:
return 'claude_session_id_own'
@property
def ready_tokens(self) -> str:
return 'Anthropic|Assistant|Chat|Welcome'
@property
def exit_key(self) -> str:
return '/exit'
@property
def delegate_agent_key(self) -> str:
return 'claude-code'
@property
def identity_cache_fields(self) -> tuple:
return ('session_id', 'session_jsonl', 'session_size_bytes', 'session_lines')
@property
def input_prompt(self) -> str:
return ''
@@ -22,3 +39,83 @@ class ClaudeAgentAdapter(BaseAgentAdapter):
@property
def input_rule_pattern(self) -> str:
return '{10,}'
def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
return f"{ctx.claude_dir}/{ctx.ws_key}/{uuid}.jsonl"
def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
path = self.artifact_path(uuid, ctx)
if not os.path.exists(path):
return False
if ctx.epoch and os.path.getmtime(path) < ctx.epoch:
return False
try:
valid_session = False
found_cwd = None
with open(path) as f:
for _ in range(50):
line = f.readline()
if not line:
break
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
if payload.get("sessionId") == uuid:
valid_session = True
if payload.get("cwd"):
found_cwd = payload.get("cwd")
break
except Exception:
pass
if not valid_session:
return False
if found_cwd and workspace_key(found_cwd) != workspace_key(ctx.cwd):
return False
except Exception:
return False
return True
def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> list:
jsonl = self.artifact_path(uuid, ctx)
if os.path.exists(jsonl):
os.remove(jsonl)
return [jsonl]
return []
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
if use_wrapper or not session_uuid:
return f"{binary} --dangerously-skip-permissions"
return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"
def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
if materialized and session_uuid:
return f"{binary} --dangerously-skip-permissions -r {session_uuid}"
elif session_uuid:
return f"{binary} --dangerously-skip-permissions --session-id {session_uuid}"
return f"{binary} --dangerously-skip-permissions"
def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
if run_cmd is not None:
try:
rc, stdout, stderr = run_cmd(['claude', 'auth', 'status'])
return rc == 0 and ('"loggedIn":true' in (stdout or '').replace(' ', ''))
except Exception:
return False
try:
res = subprocess.run(['claude', 'auth', 'status'], capture_output=True, text=True)
return res.returncode == 0 and ('"loggedIn":true' in (res.stdout or '').replace(' ', ''))
except Exception:
return False
def discover(self, ctx: DiscoveryContext) -> list:
key = workspace_key(ctx.workspace)
proj = f"{ctx.claude_dir}/{key}"
candidates = []
if os.path.isdir(proj):
for j in sorted(glob.glob(f"{proj}/*.jsonl"), key=os.path.getmtime, reverse=True):
cand = os.path.basename(j)[:-6]
if cand and self.verify_artifact(cand, ctx):
candidates.append(cand)
return candidates