Godopu
17edf90676
- Separate TUI dialog detection into blocking gate layer (_MAM_MODAL_TOKENS) and recovery layer (_MAM_HINT_TOKENS) to eliminate idle dialog starvation - Add modal_tokens contract across agent adapters (claude, cline) with dynamic scope facts resolution in send_keys_safe - Implement fail-closed resolution for ambiguous same-kind panes (R-1) and enforce WORKSPACE_ROOT export (R-2) - Preserve session on TUI readiness timeout when PID is alive for diagnostics - Add comprehensive test suite in tests/test_c1_tui_readiness.py and adapter contract assertions in test_a4
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
import os, json, shutil, glob
|
||
from typing import Optional, Any
|
||
from lib_py.agents.base import BaseAgentAdapter, DiscoveryContext
|
||
from lib_py.verify_session import workspace_key
|
||
|
||
class ClineAgentAdapter(BaseAgentAdapter):
|
||
@property
|
||
def name(self) -> str:
|
||
return 'cline'
|
||
|
||
@property
|
||
def own_key(self) -> str:
|
||
return 'cline_conversation_id_own'
|
||
|
||
@property
|
||
def ready_tokens(self) -> str:
|
||
return 'Cline|history|Chat|What can I do|slash commands'
|
||
|
||
@property
|
||
def strong_ready_tokens(self) -> str:
|
||
return 'Cline|What can I do|slash commands'
|
||
|
||
@property
|
||
def weak_ready_tokens(self) -> str:
|
||
return 'history|Chat'
|
||
|
||
@property
|
||
def modal_tokens(self) -> Optional[str]:
|
||
return 'Select API Provider|Enter API Key|Select an API provider|API Provider|Cline API key'
|
||
|
||
@property
|
||
def exit_key(self) -> str:
|
||
return '/exit'
|
||
|
||
@property
|
||
def delegate_agent_key(self) -> str:
|
||
return 'cline-agent'
|
||
|
||
@property
|
||
def identity_cache_fields(self) -> tuple:
|
||
return ('session_id',)
|
||
|
||
@property
|
||
def input_prompt(self) -> str:
|
||
return '❯'
|
||
|
||
@property
|
||
def input_placeholder(self) -> str:
|
||
return 'Ask anything...'
|
||
|
||
@property
|
||
def input_rule_pattern(self) -> str:
|
||
return '─{10,}'
|
||
|
||
def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
|
||
return f"{ctx.home_dir}/.cline/data/sessions/{uuid}/{uuid}.json"
|
||
|
||
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:
|
||
with open(path) as f:
|
||
sdata = json.load(f)
|
||
if sdata.get("session_id") != uuid:
|
||
return False
|
||
found_cwd = sdata.get("cwd") or sdata.get("workspace_root")
|
||
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:
|
||
purged = []
|
||
sessions_dir = f"{ctx.home_dir}/.cline/data/sessions/{uuid}"
|
||
if os.path.isdir(sessions_dir):
|
||
shutil.rmtree(sessions_dir)
|
||
purged.append(sessions_dir)
|
||
return purged
|
||
|
||
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
|
||
return f"{binary} -i"
|
||
|
||
def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
|
||
if materialized and session_uuid:
|
||
return f"{binary} -i --id {session_uuid}"
|
||
return f"{binary} -i"
|
||
|
||
def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
|
||
return True
|
||
|
||
def discover(self, ctx: DiscoveryContext) -> list:
|
||
sessions_dir = f"{ctx.home_dir}/.cline/data/sessions"
|
||
if os.path.isdir(sessions_dir):
|
||
files = []
|
||
for folder in glob.glob(f"{sessions_dir}/*"):
|
||
if os.path.isdir(folder):
|
||
fn = os.path.basename(folder)
|
||
jf = f"{folder}/{fn}.json"
|
||
if os.path.exists(jf):
|
||
files.append(jf)
|
||
files.sort(key=os.path.getmtime, reverse=True)
|
||
candidates = []
|
||
for j in files:
|
||
cand = os.path.basename(j)[:-5]
|
||
if cand and self.verify_artifact(cand, ctx):
|
||
candidates.append(cand)
|
||
return candidates
|
||
return []
|