import configparser import json from pathlib import Path import sqlite3 import time from 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]