Something went wrong. Try again.
Where attention appears to go across agent threads: collector, web explorer, and native iOS client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183import configparserimport jsonfrom pathlib import Pathimport sqlite3import timefrom urllib.parse import urlsplit
from public_snapshot import redact
def repository(origin): if not isinstance(origin, str) or not origin: return '' if '://' in origin: parsed = urlsplit(origin) if parsed.scheme not in ('https', 'http', 'ssh', 'git'): return '' value = (parsed.hostname or '') + parsed.path elif '@' in origin and ':' in origin: value = origin.split('@', 1)[1].replace(':', '/', 1) else: return '' return redact(value.removesuffix('.git').rstrip('/'), 240)
def codex_homes(home): return [home / '.codex', home / 'Library/Application Support/Orca/codex-runtime-home/home']
def claude_roots(home): roots = [home / '.claude/projects'] accounts = home / 'Library/Application Support/Orca/claude-accounts' roots.extend(accounts.glob('*/home/projects')) return roots
def git_context(cwd): path = Path(cwd) if not cwd or not path.is_absolute(): return {} for root in (path, *path.parents): git = root / '.git' try: if git.is_file(): value = git.read_text().strip() if not value.startswith('gitdir: '): return {} git = (root / value[8:]).resolve() if not git.is_dir(): continue common = git if (git / 'commondir').exists(): common = (git / (git / 'commondir').read_text().strip()).resolve() config = configparser.RawConfigParser(strict=False) config.read(common / 'config') origin = config.get('remote "origin"', 'url', fallback='') head = (git / 'HEAD').read_text().strip() return {'repo': repository(origin), 'branch': head.removeprefix('ref: refs/heads/') if head.startswith('ref: refs/heads/') else '', 'project': root.name, 'repo_source': 'Git origin'} except (OSError, ValueError, configparser.Error): continue return {}
class SessionContext: def __init__(self, home=None): self.home = home or Path.home() self.checked = 0 self.cache = {} self.worktrees = {} self.errors = []
def refresh(self): if time.monotonic() - self.checked < 30: return self.checked = time.monotonic() self.cache = {} self.worktrees = {} self.automations = [] self.errors = [] profiles = self.home / 'Library/Application Support/Orca/profiles' for path in profiles.glob('*/profile-state.db'): try: with sqlite3.connect(f'{path.as_uri()}?mode=ro', uri=True, timeout=1) as db: data = {k: json.loads(v) for k, v in db.execute("SELECT domain,payload FROM profile_state_documents WHERE domain IN ('repos','worktreeMeta','worktreeMetaByIdentity','automations')")} for automation in data.get('automations') or []: prompt = automation.get('prompt') if isinstance(automation, dict) else None if isinstance(prompt, str) and prompt.strip(): self.automations.append(normalized(redact(prompt, 1400))[:160]) repos = {r['id']: r for r in data.get('repos', [])} for key, meta in data.get('worktreeMeta', {}).items(): repo_id, separator, cwd = key.partition('::') if not separator or meta.get('hostId', 'local') != 'local': continue repo = repos.get(repo_id, {}) remote = repository((repo.get('gitRemoteIdentity') or {}).get('remoteUrl', '')) self.worktrees[cwd] = { 'repo': remote, 'project': repo.get('displayName') or Path(repo.get('path', '')).name, 'worktree': meta.get('displayName') or Path(cwd).name, 'host': 'Orca', 'repo_source': 'Orca repository', 'linked_pr': meta.get('linkedPR') if isinstance(meta.get('linkedPR'), int) else None, 'linked_issue': meta.get('linkedIssue') if isinstance(meta.get('linkedIssue'), int) else None, 'workspace_status': meta.get('workspaceStatus', ''), 'worktree_archived': bool(meta.get('isArchived')), } except (OSError, sqlite3.Error, ValueError, TypeError, KeyError): self.errors.append('Orca workspace metadata could not be read.')
def automated(self, text): self.refresh() text = normalized(text) return any(text.startswith(prompt) for prompt in self.automations)
def get(self, cwd, origin='', branch=''): self.refresh() if cwd not in self.cache: context = git_context(cwd) for path in (Path(cwd), *Path(cwd).parents): orca = self.worktrees.get(str(path)) if orca: context = {**orca, **context, 'project': orca.get('project') or context.get('project', '')} break self.cache[cwd] = context context = dict(self.cache[cwd]) recorded = repository(origin) if recorded: context.update(repo=recorded, repo_source='Session remote') if branch: context['branch'] = branch return {k: redact(v, 240) if isinstance(v, str) else v for k, v in context.items()}
def normalized(text): return ' '.join(text.split()).lower()
def orca_codex_rollouts(home, cutoff=0): runtime = codex_homes(home)[1] for directory in ('sessions', 'archived_sessions'): for path in (runtime / directory).glob('**/*.jsonl'): try: if path.stat().st_mtime < cutoff: continue with path.open() as stream: row = json.loads(stream.readline(1024 * 1024)) meta = row.get('payload') or {} if row.get('type') == 'session_meta' and isinstance(meta, dict) and meta.get('id'): yield path, meta, directory == 'archived_sessions' except (OSError, ValueError): continue
def recorded_workdirs(row): import re payload = row.get('payload') or {} if not isinstance(payload, dict): return [] if row.get('type') == 'turn_context' and isinstance(payload.get('cwd'), str): return [payload['cwd']] if payload.get('type') not in ('function_call', 'custom_tool_call'): return [] if payload.get('name') in ('exec_command', 'functions.exec_command'): try: arguments = json.loads(payload.get('arguments', '{}')) workdir = arguments.get('workdir') return [workdir] if isinstance(workdir, str) else [] except (ValueError, AttributeError): return [] if payload.get('name') not in ('exec', 'functions.exec'): return [] code = payload.get('input', '') if not isinstance(code, str): return [] found = [] for match in re.finditer(r'''["']?\bworkdir["']?\s*:\s*("(?:[^"\\]|\\.)*"|'[^'\\]*')''', code[:262144]): literal = match.group(1) try: value = json.loads(literal) if literal.startswith('"') else literal[1:-1] except ValueError: continue if value.startswith('/'): found.append(value) return found[:20]