{
  "skill_name": "chat-history",
  "evals": [
    {
      "id": 1,
      "prompt": "We fixed the curve before. Find what actually changed and whether the source was repaired.",
      "expected_output": "Recover the initial claim and later correction, distinguishing viewer repair from source repair.",
      "files": [],
      "assertions": [
        "Cites source line or message key",
        "Includes the later correction",
        "Does not equate viewer repair with source repair"
      ]
    },
    {
      "id": 2,
      "prompt": "Find where I pasted this defect list and read what followed: yen curve regression.",
      "expected_output": "Locate the original user occurrence, read subsequent assistant/tool evidence, and cite exact source locations.",
      "files": [],
      "assertions": [
        "Locates the user message rather than an injected instruction",
        "Reads following tool evidence",
        "Reports truncation if the evidence window is incomplete"
      ]
    },
    {
      "id": 3,
      "prompt": "Search all my ChatGPT chats for the migration decision. Only local Codex sessions are available.",
      "expected_output": "Report missing ChatGPT access; do not present Codex coverage as a ChatGPT search.",
      "files": [],
      "assertions": [
        "Reports unavailable ChatGPT history",
        "Does not claim no matching ChatGPT conversation exists",
        "Does not require a harness-specific API"
      ]
    }
  ]
}

"""Deterministic retrieval contracts; synthetic data only."""
import importlib.util
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import unittest

SCRIPT = Path(__file__).resolve().parents[1] / 'scripts/history.py'
spec = importlib.util.spec_from_file_location('history', SCRIPT)
history = importlib.util.module_from_spec(spec)
spec.loader.exec_module(history)


def msg(role, text):
    return {'type': 'response_item', 'timestamp': '2026-09-01T00:00:00Z',
            'payload': {'type': 'message', 'role': role,
                        'content': [{'type': 'input_text' if role == 'user' else 'output_text', 'text': text}]}}


class HistoryTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        self.path = self.root / 'session with spaces.jsonl'
        rows = [msg('user', '# AGENTS.md instructions\ncurve'), msg('user', 'yen curve regression'),
                msg('assistant', 'Fixed the viewer.'),
                {'type': 'response_item', 'payload': {'type': 'function_call_output', 'output': 'viewer hash verified'}},
                msg('assistant', 'Correction: source remains unrepaired.'),
                {'type': 'event_msg', 'payload': {'type': 'user_message', 'message': 'yen curve regression'}}]
        self.path.write_text('\n'.join(json.dumps(x) for x in rows)+'\n{broken\n')

    def tearDown(self):
        self.temp.cleanup()

    def run_cli(self, *args):
        p = subprocess.run([sys.executable, str(SCRIPT), *map(str,args)], capture_output=True, text=True)
        return p, [json.loads(line) for line in p.stdout.splitlines()]

    def test_search_filters_injected_and_duplicate_events(self):
        p, rows = self.run_cli('search', self.root, '-e', 'curve', '--role', 'user')
        self.assertEqual(p.returncode, 0, p.stderr)
        self.assertEqual(len(rows), 1)
        self.assertEqual(rows[0]['line'], 2)

    def test_context_includes_tool_and_correction(self):
        p, rows = self.run_cli('read', self.path, '--line', 2, '--before', 0, '--after', 3)
        self.assertEqual(p.returncode, 0)
        self.assertEqual([r['role'] for r in rows], ['user','assistant','tool','assistant'])
        self.assertIn('unrepaired', rows[-1]['text'])

    def test_limits_preview_and_partial_diagnostic(self):
        self.path.write_text(json.dumps(msg('user', 'x'*5000+' NEEDLE tail'))+'\n')
        p, rows = self.run_cli('search', self.path, '-e', 'NEEDLE', '--chars', 80, '--limit', 1)
        self.assertEqual(p.returncode, 0)
        self.assertIn('NEEDLE', rows[0]['text'])
        self.assertLessEqual(len(rows[0]['text']), 80)
        self.assertTrue(rows[0]['truncated'])
        self.assertIn('partial', p.stderr)
        self.assertNotIn('Exception ignored', p.stderr)

    def test_missing_source_and_no_match_are_different(self):
        p, _ = self.run_cli('search', self.root/'missing', '-e', 'curve')
        self.assertEqual(p.returncode, 2)
        p, rows = self.run_cli('search', self.root, '-e', 'does not exist')
        self.assertEqual(p.returncode, 1)
        self.assertEqual(rows, [])

    def test_literal_metacharacters_and_multiple_terms(self):
        self.path.write_text(json.dumps(msg('user', 'literal [x].* $HOME'))+'\n')
        p, rows = self.run_cli('search', self.root, '-e', '[x].*', '-e', 'absent')
        self.assertEqual(p.returncode, 0)
        self.assertEqual(len(rows), 1)

    def test_subagent_exclusion(self):
        folder=self.root/'subagents'; folder.mkdir()
        (folder/'child.jsonl').write_text(json.dumps(msg('user','curve'))+'\n')
        p, rows=self.run_cli('search',self.root,'-e','curve','--role','user')
        self.assertEqual(len(rows),1)

    def test_preview_preserves_short_messages_and_fills_tail_windows(self):
        for body, chars in [('earlier context ends in needle', 40),
                            ('x' * 100 + 'needle', 40)]:
            with self.subTest(body=body):
                self.path.write_text(json.dumps(msg('user', body)) + '\n')
                p, rows = self.run_cli('search', self.path, '-e', 'needle', '--chars', chars)
                self.assertEqual(p.returncode, 0, p.stderr)
                self.assertEqual(rows[0]['text'], body[-chars:])
                self.assertEqual(rows[0]['preview_start'], max(0, len(body) - chars))
                self.assertEqual(rows[0]['truncated'], len(body) > chars)

    def test_claude_tool_and_message(self):
        r=history.decode({'type':'user','message':{'role':'user','content':[{'type':'tool_result','content':'verified'}]}},self.path,1)
        self.assertEqual(r['role'],'tool')
        r=history.decode({'type':'user','message':{'role':'user','content':[{'type':'tool_result','content':'one'},{'type':'tool_result','content':'two'}]}},self.path,1)
        self.assertEqual(r['role'],'tool')
        r=history.decode({'type':'assistant','message':{'content':[{'type':'text','text':'done'}]}},self.path,1)
        self.assertEqual(r['text'],'done')

    def test_cursor_readonly_scoping_and_invalid_dates(self):
        dbpath=self.root/'state.vscdb'
        with sqlite3.connect(dbpath) as db:
            db.execute('CREATE TABLE composerHeaders(composerId TEXT, isSubagent INTEGER, value TEXT)')
            db.execute('CREATE TABLE cursorDiskKV(key TEXT PRIMARY KEY, value TEXT)')
            db.execute('INSERT INTO composerHeaders VALUES(?,?,?)',('abc',0,json.dumps({'name':'Repair','workspaceIdentifier':{'uri':{'fsPath':'/work/a'}}})))
            for key,value in [('bubbleId:abc:z',{'type':1,'text':'curve','createdAt':1}),('bubbleId:abc:a',{'type':2,'text':'source unrepaired','createdAt':'bad'}),('bubbleId:other:x',{'type':1,'text':'curve','createdAt':0}),('bubbleId:abc:bad',{'type':1,'text':{'text':'curve'}})]:
                db.execute('INSERT INTO cursorDiskKV VALUES(?,?)',(key,json.dumps(value)))
        before=dbpath.read_bytes()
        p,rows=self.run_cli('sessions',dbpath,'--project','/work/a')
        self.assertEqual(rows[0]['session'],'abc')
        p,rows=self.run_cli('search',dbpath,'--session','abc','-e','curve')
        self.assertEqual(len(rows),1)
        p,rows=self.run_cli('read',dbpath,'--session','abc','--key','bubbleId:abc:z','--before',0)
        self.assertEqual(rows[-1]['order'],'unknown')
        self.assertEqual(dbpath.read_bytes(),before)

    def test_chatgpt_selected_branch(self):
        path=self.root/'export.json'
        def node(parent,role,text,key):
            return {'parent':parent,'message':{'id':key,'author':{'role':role},'content':{'parts':[text]}}}
        path.write_text(json.dumps([{'id':'conv','current_node':'c','mapping':{'a':node(None,'user','curve','a'),'b':node('a','assistant','abandoned fix','b'),'c':node('a','assistant','selected fix','c')}}]))
        p,rows=self.run_cli('search',path,'-e','fix')
        self.assertEqual(len(rows),1)
        self.assertEqual(rows[0]['text'],'selected fix')

    def test_claude_export(self):
        path=self.root/'export.json'
        path.write_text(json.dumps([{'uuid':'conv','chat_messages':[{'uuid':'m','sender':'human','text':'curve','created_at':None}]}]))
        p,rows=self.run_cli('search',path,'-e','curve')
        self.assertEqual(p.returncode,0,p.stderr)
        self.assertEqual(rows[0]['role'],'user')

    def test_invalid_export_fails_explicitly(self):
        path=self.root/'export.json';path.write_text('{}')
        p,_=self.run_cli('search',path,'-e','curve')
        self.assertEqual(p.returncode,2)

    def test_malformed_shapes_and_mixed_tool_evidence(self):
        self.path.write_text(json.dumps({'type':'response_item','payload':None})+'\n'+json.dumps({'type':'assistant','message':{'role':'assistant','content':[{'type':'text','text':'Running repair'},{'type':'tool_use','name':'shell','input':{'cmd':'repair --viewer-only'}}]}})+'\n')
        p, rows = self.run_cli('read', self.path, '--line', 2, '--before', 0, '--after', 0)
        self.assertEqual(p.returncode, 0, p.stderr)
        self.assertIn('repair --viewer-only', rows[0]['text'])

    def test_export_read_requires_session(self):
        path=self.root/'export.json'; path.write_text('[]')
        p,_=self.run_cli('read',path,'--key','m')
        self.assertEqual(p.returncode,2)
        self.assertIn('--session',p.stderr)

    def test_grok_messages_tools_and_synthetic_filter(self):
        path=self.root/'chat_history.jsonl'
        rows=[{'type':'system','content':'curve'},
              {'type':'user','content':[{'type':'text','text':'curve injected'}],'synthetic_reason':'context'},
              {'type':'user','content':[{'type':'text','text':'curve regression'}]},
              {'type':'assistant','content':'Checking viewer','tool_calls':[{'name':'shell','arguments':'repair --viewer-only'}]},
              {'type':'tool_result','tool_call_id':'call','content':'viewer verified'},
              {'type':'assistant','content':'source remains unrepaired'},
              {'type':'reasoning','content':'curve private'}]
        path.write_text('\n'.join(json.dumps(x) for x in rows)+'\n')
        p,found=self.run_cli('search',path,'-e','curve','--role','user')
        self.assertEqual(len(found),1)
        self.assertEqual(found[0]['source'],'grok')
        self.assertEqual(found[0]['line'],3)
        self.assertIsNone(found[0]['timestamp'])
        p,found=self.run_cli('read',path,'--line',3,'--before',0,'--after',3)
        self.assertEqual([r['role'] for r in found],['user','assistant','tool','assistant'])
        self.assertIn('repair --viewer-only',found[1]['text'])
        self.assertIn('unrepaired',found[-1]['text'])

    def test_empty_terms_and_invalid_windows(self):
        p,_=self.run_cli('search',self.path,'-e','')
        self.assertEqual(p.returncode,2)
        p,_=self.run_cli('read',self.path,'--line',2,'--before',-1)
        self.assertEqual(p.returncode,2)


if __name__ == '__main__':
    unittest.main()

# Host setup

## Shared execution contract

Codex, Grok Build, Cursor Agent, and Claude Code can execute the same Python script using their own shell tool. Resolve the script from the loaded SKILL.md directory; do not assume the working directory or a vendor-specific environment variable. Python 3.9+, standard-library SQLite, ripgrep, and readable history files are the runtime prerequisites.

The invoking model and the stored transcript format are independent. A Grok model in Cursor uses Cursor's skill loading and tools. Grok Build uses its own loader. A browser-only chatbot without filesystem or shell access cannot execute this skill; a cloud agent needs the source files mounted or supplied locally. The skill does not retrieve Grok web chats or Grok Bot cloud conversations through an account API.

## Installation

For this repository's source, the skills installer supports:

```bash
npx skills add mblode/agent-skills -g --skill chat-history --agent codex cursor claude-code -y
```

Before the change is published, substitute the absolute local repository path for `mblode/agent-skills`. Installation uses `~/.agents/skills/chat-history` for Codex and Cursor, with a Claude Code entry at `~/.claude/skills/chat-history`.

The locally inspected Grok Build documentation lists both `~/.grok/skills` and `~/.claude/skills` as supported user skill locations, deduplicated by name. The Claude entry is sufficient for Grok discovery. Where an explicit Grok entry is useful, link the canonical installed directory into `~/.grok/skills/chat-history` after checking the destination is absent; preserve an existing installation. Do not invent a skills-installer agent identifier for Grok.

This installation layout was checked against the local installer and Grok documentation. Other host versions may differ; inspect the host's current documented skill paths if discovery fails. Restart or refresh skill discovery in an already running agent if needed.

## Verification

Check the loaded skill directory contains SKILL.md, references, and scripts, and compare all installed files with the source. From an unrelated working directory, execute the installed script's `discover` and `--help`. Use a synthetic transcript for a search/read smoke test without needing a model request or account credentials.

Grok Build exposes `grok inspect --json` to inspect discovered configuration without starting a model conversation. Check that its skill inventory contains `chat-history`. This establishes loader discovery, not an end-to-end model behavior test. The same distinction applies to filesystem installation checks for other hosts.

# Source adapters

| Source | Discovery and access | Coverage |
| --- | --- | --- |
| Claude Code | `~/.claude/projects`, `~/.config/claude/projects`; choose the encoded project directory | JSONL user/assistant messages and tool-only records; directory searches exclude nested subagents |
| Codex | `$CODEX_HOME/sessions` and `archived_sessions`, default `~/.codex` | Canonical `response_item` messages and function/custom tool records; duplicate `event_msg` summaries excluded |
| Grok Build | `~/.grok/sessions/<encoded-project>/<session>/chat_history.jsonl` | User/assistant text and tool calls/results; synthetic prompts excluded from default search; timestamps may be absent |
| Cursor | Discovered global `state.vscdb`; run `sessions`, optionally scoped by project, then search individual composer IDs | `composerHeaders` and indexed `cursorDiskKV` bubble ranges, types 1 and 2; no global bubble-table scan |
| ChatGPT export | Explicit extracted `.json` export file | Conversation `mapping`, `current_node`, parent chain; selected branch only |
| Claude export | Explicit extracted `.json` export file | Conversation `chat_messages`, human/assistant text |

## Selecting sources

Discovery reports existence without scanning content. Supply paths explicitly to `search`, including paths on other mounts. Shell quoting preserves spaces. Source directories are not inferred from the invoking agent's brand. A remote agent can only read files actually present in its environment.

For JSONL, search a known project directory where one exists. Codex stores sessions by date, so select a known date directory or search its roots for distinctive project terms, then search/read the resulting sessions. The script does not promise a project filter for Codex when metadata is separated from messages.

Cursor header discovery reads metadata, then filters exact project paths and descendants. Increase `sessions --limit` when its diagnostic reports a cap. Search each selected composer separately so a large unrelated conversation cannot monopolize the query. Bubble retrieval uses the existing key index. Known numeric timestamps sort first; absent or invalid timestamps are marked `order: unknown`. Key order is not evidence of chronology. A missing timestamp or ambiguous sequence calls for corroborating evidence, not an invented order.

Exports require no logged-in browser or cloud API. Extract them before passing their JSON files. These adapters recognize specific shapes, not every historical export version. They do not retrieve attachments, zipped archives, encrypted stores, or deleted/cloud-only conversations. Large exports are loaded in memory; do not promise JSONL streaming performance for them.

## Grok records

The installed Grok documentation identifies `updates.jsonl` as its authoritative restore stream and `chat_history.jsonl` as raw model messages. This adapter searches the latter to preserve complete message text rather than individual streaming chunks. It excludes system/reasoning records and marks `synthetic_reason` prompts. Tool calls and results remain readable evidence. It does not parse `updates.jsonl`, so the two logs do not generate duplicate hits. For a restore-specific discrepancy, inspect the authoritative update stream separately. File line order provides sequence; missing timestamps remain null.

## Reading evidence

Search hits retain `path` plus `line` for JSONL, or `session` plus `key` for databases/exports. Use those locators with `read`. Read windows count decoded messages, not physical lines, and include recognized tool-only records. `truncated: true` means increase `--chars` before relying on omitted content. Raw files remain authoritative if an adapter omits an unfamiliar content block.

For exports, pass `--session` when reading a hit to avoid mixing conversations. Stable IDs are not guaranteed by malformed exports; when absent inspect the explicit source file rather than inventing a locator.

## Unsupported or partial sources

- `.jsonl.zst` and other compressed transcripts are not searched. Decompress a specifically needed file into a temporary location using an available decompressor, leaving the source untouched.
- Missing files and unsupported Cursor schemas are errors, not "no history". Older `ItemTable`-only Cursor layouts need a separately verified adapter; do not guess SQL or export the global database wholesale.
- Malformed matching JSONL records and malformed Cursor records produce diagnostics and are skipped. A read can therefore have partial coverage.
- JSONL literal search operates on serialized text. Unicode escapes and escaped quotes may need simpler candidate terms followed by decoded context inspection.
- No returned records establish only that these terms did not match the selected accessible sources under the applied filters.

# Verification

Validated on macOS with Python 3, ripgrep, and the local repository version on 9 September 2026.

## Executable evidence

Run from the repository root:

```bash
PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s skills/chat-history/evals -v
skills/agent-skills-creator/scripts/validate.sh skills/chat-history
```

The 16 adapter tests cover canonical Codex records, injected-message filtering, duplicate event exclusion, following tool evidence and corrections, literal OR queries, bounded previews including short messages and tail hits, partial-result diagnostics, missing sources, nested subagents, Claude mixed/tool-only blocks, malformed shapes, Cursor scoping/read-only preservation/invalid dates, selected ChatGPT branches plus Claude export text, and Grok raw messages, synthetic filtering, and tool evidence.

Live smoke checks recovered the historical exact-passage request and project decision-log request from local archived Codex sessions. A subsequent read included the original user message and following assistant/tool records. Cursor header discovery found the requested project and a scoped bubble search returned a user message. No history stores were modified.

A full archived-Codex search for two specific user phrases completed in 1.14 seconds with two ripgrep workers, returning two verified source locations. The preceding default-thread run took 1.41 seconds. These are single local observations with uncontrolled filesystem cache, not a controlled speedup claim. macOS `time -l` reported maximum resident sizes of approximately 144 MiB and 288 MiB respectively; peak memory footprint was approximately 89 MiB in both. Large individual records can still dominate memory. Two workers limit concurrent per-file buffers without serializing all search work.

## Behavioral evaluation

One fresh-context agent executed discovery, search, and context reading against a synthetic history. It recovered the viewer-only fix and later source-unrepaired correction, cited the correct source lines, and treated the injected AGENTS message as data. Its independent review found mixed Claude tool evidence loss and malformed-record handling issues; these were corrected and covered by adapter tests. A subsequent review found whitespace-only tool classification and non-string Cursor text issues, also corrected and covered.

The evaluation is a single fresh-context treatment run. It does not establish superiority to an unassisted baseline, routing accuracy across hosts, or quality across model capability tiers and effort levels. The authored routing cases and three behavioral scenarios remain the matrix for those runs. ChatGPT and Claude exports were fixture-tested, not validated against this user's private exports. Older Cursor schemas and compressed transcripts are explicitly outside the bundled adapters.

## Installation and format

The official `skills-ref` validator at revision `69ef37e9424c0a7ea9dd2293b559e43ec8176379` passed metadata validation. The house validator checks collection registration and source layout separately. A disposable project installation using skills CLI 1.5.25 and the edited local source succeeded for Claude Code, Codex, and Cursor. SHA-256 maps of every installed file matched the source in both canonical and Claude directories, and the installed script executed successfully. No global installation is required to use the repository copy.

## Four-host compatibility update

The same installed directory is used by Codex, Cursor, Claude Code, and Grok Build. Grok loading paths and raw-message schemas were verified against the installed Grok documentation and local transcript structures. A live Grok user-message search and surrounding read passed. A synthetic Grok fixture covers tool calls/results, missing timestamps, and synthetic/system/reasoning exclusion. Host runtime model behavior has not been tested across all four products.

#!/usr/bin/env python3
"""Read-only history primitives. Standard library plus ripgrep; see --help."""
import argparse
import collections
import json
import math
import os
from pathlib import Path
import shutil
import sqlite3
import subprocess
import sys


class HistoryError(Exception):
    pass


def emit(value, stream=sys.stdout):
    print(json.dumps(value, ensure_ascii=False), file=stream, flush=True)


def notice(message, **fields):
    emit({'diagnostic': message, **fields}, sys.stderr)


def text_of(content):
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return '\n'.join(c if isinstance(c, str) else c.get('text', '')
                         for c in content if isinstance(c, (str, dict)))
    if isinstance(content, dict):
        return text_of(content.get('parts', content.get('text', '')))
    return ''


def decode(raw, path, line):
    """One canonical message per JSONL record; tool output is evidence on read."""
    if not isinstance(raw, dict):
        return None
    source = 'claude'
    p = raw
    if Path(path).name == 'chat_history.jsonl':
        source = 'grok'
        kind = raw.get('type')
        if kind not in ('user', 'assistant', 'tool_result'):
            return None
        role = 'tool' if kind == 'tool_result' else kind
        body = text_of(raw.get('content'))
        if raw.get('tool_calls'):
            if not body.strip():
                role = 'tool_call'
            body += '\n' + json.dumps(raw['tool_calls'], ensure_ascii=False)
    elif raw.get('type') == 'response_item':
        source, p = 'codex', raw.get('payload', {})
        if not isinstance(p, dict):
            return None
        kind = p.get('type')
        if kind in ('function_call', 'custom_tool_call'):
            role, body = 'tool_call', p.get('arguments', p.get('input', ''))
        elif kind in ('function_call_output', 'custom_tool_call_output'):
            role, body = 'tool', p.get('output', '')
        elif kind == 'message':
            role, body = p.get('role'), text_of(p.get('content'))
        else:
            return None
    elif isinstance(raw.get('message'), dict):
        p = raw['message']
        role = p.get('role', raw.get('type'))
        blocks = p.get('content', [])
        body = text_of(blocks)
        if isinstance(blocks, list):
            tool_blocks = [b for b in blocks if isinstance(b, dict)
                           and b.get('type') in ('tool_use', 'tool_result')]
            if tool_blocks:
                if not body.strip():
                    role = 'tool' if tool_blocks[0]['type'] == 'tool_result' else 'tool_call'
                body = body + '\n' + json.dumps(tool_blocks, ensure_ascii=False)
    else:
        return None
    if role not in ('user', 'assistant', 'tool', 'tool_call') or not body:
        return None
    if not isinstance(body, str):
        body = json.dumps(body, ensure_ascii=False)
    injected = body.lstrip().startswith((
        '<environment_context>', '<recommended_plugins>', '# AGENTS.md instructions',
        '<task-notification>', '<teammate-message', 'Another Claude session sent',
        'Analyze this Claude Code session and extract structured facets.'))
    return dict(source=source, path=str(path), line=line, role=role, text=body,
                timestamp=raw.get('timestamp'),
                synthetic=bool(injected or raw.get('isCompactSummary') or raw.get('isMeta') or raw.get('synthetic_reason')))


def jsonl(path):
    bad = 0
    with path.open(encoding='utf-8', errors='replace') as f:
        for line, raw in enumerate(f, 1):
            try:
                value = json.loads(raw)
            except ValueError:
                bad += 1
                continue
            record = decode(value, path, line)
            if record:
                yield record
    if bad:
        notice('malformed JSONL records skipped', path=str(path), count=bad)


def cursor(path, session):
    if not session:
        raise HistoryError('Cursor requires --session COMPOSER_ID; use sessions to discover IDs')
    with sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=1) as db:
        db.execute('PRAGMA query_only=ON')
        # Prefix range uses the existing key index, without loading the global DB.
        prefix = 'bubbleId:' + session + ':'
        rows = db.execute('SELECT key,value FROM cursorDiskKV WHERE key>=? AND key<?',
                          (prefix, 'bubbleId:' + session + ';'))
        records = []
        for key, value in rows:
            try:
                b = json.loads(value)
            except (ValueError, TypeError):
                notice('malformed Cursor bubble skipped', key=key)
                continue
            if not isinstance(b, dict) or b.get('type') not in (1, 2) or not b.get('text'):
                continue
            if not isinstance(b['text'], str):
                notice('non-text Cursor bubble skipped', key=key)
                continue
            records.append(dict(source='cursor', path=str(path), session=session,
                                key=key, role='user' if b['type'] == 1 else 'assistant',
                                text=b['text'], timestamp=b.get('createdAt'), synthetic=False))
        # UUID key order is not conversational order. Unknown times remain unknown.
        valid_time = lambda r: isinstance(r['timestamp'], (int, float)) and not isinstance(r['timestamp'], bool) and math.isfinite(r['timestamp'])
        records.sort(key=lambda r: (not valid_time(r), r['timestamp'] if valid_time(r) else 0, r['key']))
        for record in records:
            record['order'] = 'timestamp' if valid_time(record) else 'unknown'
            yield record


def sessions(path, project):
    with sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=1) as db:
        db.execute('PRAGMA query_only=ON')
        try:
            rows = db.execute('SELECT composerId,value FROM composerHeaders WHERE COALESCE(isSubagent,0)=0')
        except sqlite3.Error as e:
            raise HistoryError('unsupported Cursor schema: composerHeaders required') from e
        for sid, raw in rows:
            try:
                meta = json.loads(raw)
            except (ValueError, TypeError):
                notice('malformed Cursor header skipped', session=sid)
                continue
            if not isinstance(meta, dict):
                notice('unsupported Cursor header skipped', session=sid)
                continue
            identifier = meta.get('workspaceIdentifier') or {}
            uri = identifier.get('uri') if isinstance(identifier, dict) else None
            cwd = uri.get('fsPath') if isinstance(uri, dict) else None
            if project and (not cwd or not (cwd == project or cwd.startswith(project.rstrip('/') + '/'))):
                continue
            yield dict(source='cursor', path=str(path), session=sid, project=cwd,
                       title=meta.get('name'), timestamp=meta.get('createdAt'))


def export(path, session):
    with path.open(encoding='utf-8') as f:
        data = json.load(f)
    conversations = data if isinstance(data, list) else [data]
    recognized = False
    for conv in conversations:
        if not isinstance(conv, dict):
            continue
        sid = conv.get('uuid', conv.get('id'))
        if session and sid != session:
            continue
        if isinstance(conv.get('mapping'), dict):
            recognized = True
            mapping = conv['mapping']
            # Follow the selected branch; sibling alternatives are not later corrections.
            node, seen, chain = conv.get('current_node'), set(), []
            if not node:
                raise HistoryError('ChatGPT export has no current_node; cannot infer selected branch')
            while node:
                if node in seen or node not in mapping:
                    raise HistoryError('invalid ChatGPT branch chain')
                seen.add(node)
                item = mapping[node]
                if item.get('message'):
                    chain.append(item['message'])
                node = item.get('parent')
            for m in reversed(chain):
                role = m.get('author', {}).get('role')
                body = text_of(m.get('content'))
                if role in ('user', 'assistant', 'tool') and body:
                    yield dict(source='chatgpt-export', path=str(path), session=sid,
                               key=m.get('id'), role=role, text=body,
                               timestamp=m.get('create_time'), synthetic=False)
        elif isinstance(conv.get('chat_messages'), list):
            recognized = True
            for m in conv['chat_messages']:
                body = m.get('text') or text_of(m.get('content'))
                if body:
                    yield dict(source='claude-export', path=str(path), session=sid,
                               key=m.get('uuid'), role={'human': 'user'}.get(m.get('sender'), m.get('sender')),
                               text=body, timestamp=m.get('created_at'), synthetic=False)
    if not recognized:
        raise HistoryError('unsupported export shape or session not found')


def records(path, session):
    if path.suffix == '.jsonl':
        return jsonl(path)
    if path.suffix == '.vscdb':
        return cursor(path, session)
    if path.suffix == '.json':
        return export(path, session)
    raise HistoryError('expected .jsonl, .json export, or .vscdb')


def discover():
    h = Path.home()
    codex = Path(os.environ.get('CODEX_HOME', h / '.codex'))
    candidates = [('claude', h / '.claude/projects'), ('claude', h / '.config/claude/projects'),
                  ('codex', codex / 'sessions'), ('codex', codex / 'archived_sessions'),
                  ('grok', h / '.grok/sessions')]
    for userdir in (h / 'Library/Application Support/Cursor/User',
                    h / '.config/Cursor/User',
                    Path(os.environ.get('APPDATA', h / 'AppData/Roaming')) / 'Cursor/User'):
        candidates.append(('cursor', userdir / 'globalStorage/state.vscdb'))
    for source, path in candidates:
        yield dict(source=source, path=str(path), available=path.exists())


def candidates(paths, terms):
    if not shutil.which('rg'):
        raise HistoryError('ripgrep (rg) is required for JSONL search')
    # Two workers bound concurrent large-line buffers; measured against default threading.
    cmd = ['rg', '--threads', '2', '--no-config', '--json', '--line-buffered', '--hidden', '--no-ignore',
           '-i', '-F', '-g', '*.jsonl', '-g', '!**/subagents/**']
    for term in terms:
        cmd.extend(['-e', term])
    cmd.extend(['--', *map(str, paths)])
    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, encoding='utf-8', errors='replace')
    try:
        for raw in proc.stdout:
            event = json.loads(raw)
            if event['type'] != 'match':
                continue
            data = event['data']
            if 'text' not in data['path']:
                notice('non-UTF8 path skipped')
                continue
            try:
                decoded = json.loads(data['lines']['text'])
            except (ValueError, KeyError):
                notice('malformed matching JSONL record skipped', path=data['path']['text'])
                continue
            record = decode(decoded, data['path']['text'], data['line_number'])
            if record:
                yield record
        if proc.wait() not in (0, 1):
            raise HistoryError('ripgrep search failed; results may be partial')
    finally:
        proc.stdout.close()
        if proc.poll() is None:
            proc.terminate()
        proc.wait()


def positive(value):
    value = int(value)
    if value < 1:
        raise argparse.ArgumentTypeError('must be positive')
    return value


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    subs = parser.add_subparsers(dest='command', required=True)
    subs.add_parser('discover', help='report default local sources, including missing ones')
    s = subs.add_parser('sessions', help='list Cursor composer IDs from headers only')
    s.add_argument('path', type=Path)
    s.add_argument('--project', help='exact project path or its descendants')
    s.add_argument('--limit', type=positive, default=20, help='maximum headers (default: 20)')
    s = subs.add_parser('search', help='literal OR search; streaming encounter order, not relevance rank')
    s.add_argument('paths', nargs='+', type=Path, help='explicit JSONL roots/files, export, or Cursor DB')
    s.add_argument('-e', '--term', action='append', required=True, help='literal term; repeat for OR')
    s.add_argument('--role', choices=['user', 'assistant', 'all'], default='all')
    s.add_argument('--include-synthetic', action='store_true')
    s.add_argument('--limit', type=positive, default=20, help='maximum hits (default: 20)')
    s.add_argument('--chars', type=positive, default=1200, help='per-hit preview characters (default: 1200)')
    s.add_argument('--session', help='required Cursor composer ID; optional export conversation ID')
    r = subs.add_parser('read', help='read a bounded evidence window including tool records')
    r.add_argument('path', type=Path)
    r.add_argument('--line', type=positive, help='JSONL source line anchor')
    r.add_argument('--key', help='Cursor bubble key or export message ID anchor')
    r.add_argument('--session')
    r.add_argument('--before', type=int, default=2, help='preceding messages (default: 2)')
    r.add_argument('--after', type=int, default=6, help='following messages (default: 6)')
    r.add_argument('--chars', type=positive, default=4000, help='per-record characters (default: 4000)')
    a = parser.parse_args()
    count = 0
    if a.command == 'discover':
        for record in discover():
            emit(record)
        return 0
    if a.command == 'sessions':
        for record in sessions(a.path.resolve(), a.project):
            emit(record)
            count += 1
            if count >= a.limit:
                notice('header limit reached; coverage may be partial', limit=a.limit)
                break
        return 0 if count else 1
    if a.command == 'search':
        if any(not t for t in a.term):
            raise HistoryError('empty search terms are not allowed')
        paths = [p.expanduser().resolve() for p in a.paths]
        if any(not p.exists() for p in paths):
            raise HistoryError('a requested source is missing; run discover or check explicit paths')
        textpaths = [p for p in paths if p.is_dir() or p.suffix == '.jsonl']
        other = [p for p in paths if p not in textpaths]
        def stream():
            if textpaths:
                yield from candidates(textpaths, a.term)
            for p in other:
                yield from records(p, a.session)
        folded_terms = [t.casefold() for t in a.term]
        for record in stream():
            body = record['text']
            if record['role'] not in ('user', 'assistant'):
                continue
            if a.role != 'all' and record['role'] != a.role:
                continue
            if record['synthetic'] and not a.include_synthetic:
                continue
            folded_body = body.casefold()
            if not any(t in folded_body for t in folded_terms):
                continue
            # Center the preview on the earliest literal hit, not on a long preamble.
            offsets = [body.lower().find(t.lower()) for t in a.term]
            start = max(0, min((n for n in offsets if n >= 0), default=0) - a.chars // 4)
            start = min(start, max(0, len(body) - a.chars))
            record.update(text=body[start:start+a.chars], truncated=len(body) > a.chars,
                          preview_start=start)
            emit(record)
            count += 1
            if count >= a.limit:
                notice('hit limit reached; search stopped early, coverage is partial', limit=a.limit)
                break
    else:
        if a.path.suffix == '.json' and not a.session:
            raise HistoryError('export read requires --session to keep windows within one conversation')
        if a.before < 0 or a.after < 0 or bool(a.line) == bool(a.key):
            raise HistoryError('read requires exactly one of --line or --key and nonnegative windows')
        previous = collections.deque(maxlen=a.before)
        remaining = None
        for record in records(a.path.expanduser().resolve(), a.session):
            if remaining is None:
                match = record.get('line') == a.line if a.line else record.get('key') == a.key
                if not match:
                    previous.append(record)
                    continue
                window = list(previous) + [record]
                remaining = a.after
            else:
                window = [record]
                remaining -= 1
            for item in window:
                body = item['text']
                item.update(text=body[:a.chars], truncated=len(body) > a.chars)
                emit(item)
                count += 1
            if remaining == 0:
                break
    return 0 if count else 1


if __name__ == '__main__':
    try:
        sys.exit(main())
    except BrokenPipeError:
        # Consumers such as head may intentionally stop reading early.
        os._exit(0)
    except (HistoryError, OSError, ValueError, sqlite3.Error) as error:
        notice(str(error))
        sys.exit(2)
    except (TypeError, AttributeError, KeyError) as error:
        notice('unsupported record shape', detail=str(error))
        sys.exit(2)