Stream Markup Capture

Inline markup tags in Claude responses are extracted, persisted, and routed without interrupting conversational flow. This lets silicon-Zack embed structured data (observations, follow-ups, things to file) directly in responses. Tags are stripped before delivery so bio-Zack never sees them. Two paths feed the same daily capture log: a real-time path for Telegram responses and a batch scanner for Claude Code CLI sessions.

Available Tags

Six tag names are recognized. Five are capture tags (their content is filed and routed); the sixth, <telegram-message>, is a delivery-control tag that decides what (if anything) gets sent to the user.

Tag	Purpose	Routed To
<file-this>	Facts, events, context worth remembering	data/scratch/daily.md (via /sleep)
<observation>	Durable identity observations	data/identity/observations.jsonl (via /sleep)
<follow-up>	Commitments, things to track	data/assistant/commitments.json (via /sleep)
<note-to-self>	Internal reminders for future sessions	data/scratch/daily.md (via /sleep)
<ask-next-session>	Questions to ask bio-Zack next time	data/stream-captures/next-session-questions.json (routed immediately)
<telegram-message>	Delivery control: explicit message body to send	Telegram (output only — never persisted as a capture)

The capture tags live in TAG_NAMES (a frozenset in telegram_integration.py); the regex TAG_PATTERN is built as an alternation over all six. A reimplementer reading that alternation will see telegram-message in the set — it shares the extraction machinery but is handled as an output directive, not stored.

Note: <telegram-message> exists because not every Claude turn should produce a user-visible message. For event/proactive turns (no human waiting), no tag means silence — _extract_telegram_messages() returns None and the caller stays quiet. For user-initiated turns, bare text is the fallback when no tag is present.

Architecture Overview

Claude Response (with inline tags)
       |
       v
+------+------------------+
| Tag Extraction          |  --> Regex: <tag>content</tag>  (TAG_PATTERN)
| (telegram_integration)  |      Returns clean text + captures
+------+------------------+
       |
       +----------+-------------------+
       |          |                   |
       v          v                   v
  Tag Captures   Fallback Sweep    Clean Text
  (explicit)     (13 regex pats)   (tags stripped)
       |          |                   |
       +----+-----+                   v
            |                    Delivered to
            v                    bio-Zack
   +--------+---------+
   | persist_captures()|  --> data/stream-captures/YYYY-MM-DD.jsonl
   +--------+---------+
            |
            v (if ask-next-session)
   +--------+---------+
   | next_session.py  |  --> data/stream-captures/next-session-questions.json
   +------------------+


CLI Sessions (scanned every 15 min by the consolidation loop)
       |
       v
+------+------------------+
| cli_scanner.py          |  --> Scans ~/.claude/projects/.../*.jsonl
| (24h first-run lookback)|      Tracks last scan timestamp + hashes
+------+------------------+
       |
       v
   +---+---------------+
   | write_captures()   |  --> data/stream-captures/YYYY-MM-DD.jsonl
   | (deduped by hash)  |      source: "cli_session"
   +--------------------+

Two Processing Paths

1. Telegram Real-Time

Tags are extracted from every Claude response before delivery to the user, inline in the parallel-processor pipeline (scripts/telegram_parallel/parallel_processor.py).

Flow (numbered as in process_request):

1. Claude responds with text containing markup tags.
2. Step 3 of the processor runs the capture handling: _extract_telegram_messages() pulls any <telegram-message> body (used as the outgoing text if present), then extract_and_strip_tags() removes the remaining inline tags from the visible text.
3. fallback_sweep.sweep_response() runs pattern matching on the cleaned text for untagged capturable content, persisting matches.
4. <ask-next-session> captures from the sweep are routed to the question queue via next_session.add_question().
5. Clean text (tags stripped) is delivered to bio-Zack (step 4, with image support).

Note: In the parallel-processor path, the inline capture tags are stripped from the visible text but not themselves persisted — only the fallback sweep's matches are written (with source: "fallback_sweep"). The fuller telegram_integration.process_response() helper does persist inline tags (as telegram_realtime) alongside fallback captures (telegram_fallback); it's the reusable building block. The live Telegram daemon currently wires the sweep directly in step 3.

Safety: The fallback sweep is wrapped in try/except; on any failure it logs Fallback sweep failed and delivery proceeds with the cleaned text. process_response() goes further — wrapping the whole pipeline so that on any error the original response is returned unmodified. Capture failures never block message delivery.

Streaming-aware extraction

Telegram responses arrive as a token stream, so tags can be split across chunk boundaries (e.g. <file-th at the end of one chunk, is>... at the start of the next). telegram_integration.extract_complete_tags() handles incremental extraction:

• Matches all complete <tag>...</tag> pairs in the current buffer.
• Keeps unresolved open tags in the returned remainder (via OPEN_TAG_PATTERN) so they can complete on a later chunk.
• Keeps a trailing partial tag — anything in the last PARTIAL_TAG_WINDOW = 64 characters that could be the start of a known tag (<file-th, </obser, …) — in the remainder.
• On flush=True, returns everything and clears the remainder.

The non-streaming whole-text path is extract_and_strip_tags(), which extracts captures and returns cleaned text in one pass.

2. CLI Session Scanner

cli_scanner.scan_cli_sessions() reads Claude Code session transcripts from the local machine and extracts the same five capture tags. This catches tags from interactive CLI sessions that never go through Telegram.

When it runs: continuously, as a sub-task of the micro-consolidation loop. MicroConsolidationLoop ticks every 60s during active hours (7am–11pm PT) and debounces each sub-task; the CLI scan fires on a CLI_SCAN_INTERVAL = 900 (15-minute) cadence. It is not gated on /sleep — /sleep only reads the resulting captures during reflection. A manual run_once() also triggers a scan (used by the scheduler API).

Flow:

1. Load scan state (last_scan_ts, processed_hashes).
2. Discover session files modified since the last scan (first run / missing state falls back to MAX_FIRST_RUN_LOOKBACK_HOURS = 24).
3. For each session, parse JSONL lines, take only type == "assistant" messages, and extract tags from their text content blocks.
4. Dedup by content hash against today's + yesterday's capture files and the rolling state-hash window.
5. Write survivors with source: "cli_session", partitioned by the capture's own timestamp.
6. Update scan state.

Session files: ~/.claude/projects/<project-slug>/<uuid>.jsonl. The vita project slug is auto-derived from the repo path (each / becomes -, with a leading -; e.g. /home/zack/work/vita → -home-zack-work-vita), overridable via the VITA_PROJECT_SLUG / CLAUDE_PROJECTS_DIR env vars. agent-*.jsonl sub-session files and empty files are skipped.

State tracking: data/stream-captures/.cli-scan-state.json stores last_scan_ts and a rolling window of the last 1000 processed content hashes (all_hashes[-1000:]).

Note: The CLI scanner runs no fallback sweep — it only extracts explicit tags. The regex/LLM-style fallback is exclusive to the Telegram real-time path.

Fallback Sweep

A secondary safety net that catches capturable content the model didn't explicitly tag. Runs on the cleaned text after tag extraction in the Telegram path.

13 regex patterns across 5 tag types (ALL_PATTERNS in fallback_sweep.py):

Tag Type	# Patterns	Pattern Examples
file-this	4	"worth remembering …", "he mentioned …", "meeting tomorrow …", "[Name] works at …"
observation	2	"I notice …", "he consistently / prefers / values …"
follow-up	2	"I'll follow up on …", "I'll remind / schedule / draft …"
note-to-self	3	"note to self: …", "correction: …", "I'll keep that in mind …"
ask-next-session	2	"next time I'll ask …", "I'm curious about …"

Each PatternDef carries a compiled regex, target tag, content capture group, a confidence score, and a require_substance flag.

Mechanisms beyond plain matching:

• Specificity ordering + span suppression — patterns run most-specific-first. detect_patterns() records the character span of each accepted match; a later pattern whose span overlaps an already-claimed range is dropped (_overlaps_any()). Earlier, higher-priority patterns win the text.
• Substance check — require_substance (default True) rejects matches whose captured content is shorter than 20 characters. A few short-phrase patterns (e.g. keep_in_mind) opt out.
• Meta-noise filter — _is_meta_noise() rejects content that's about the system itself (tags, CLAUDE.md, "stream capture", "fallback sweep", "pipeline", …).
• Tool-narration filter — _is_tool_narration() drops Claude narrating its own current tool use ("Found it", file paths, line numbers, mid-word fragments).

Dedup: Fallback captures are checked against the inline captures already extracted from the same response, using substring matching and a >0.70 token-overlap threshold (deduplicate()). Self-dedup among fallback captures uses a >0.60 overlap threshold, keeping the higher-confidence (then longer) match (_self_deduplicate()).

Deduplication

Content is deduplicated at multiple levels:

1. Content hash: SHA-256 of normalized text (lowered, punctuation stripped, whitespace collapsed). The CLI scanner checks against today's + yesterday's capture files and the rolling state window; the Telegram persister stores a 16-char hash prefix per record.
2. Inline vs fallback: the fallback sweep skips content already captured by explicit tags (substring + >0.70 token overlap).
3. Source tagging: every record carries a source field marking which path produced it:

source value	Produced by
telegram_realtime	Inline tags persisted by telegram_integration.process_response()
telegram_fallback	Fallback-sweep matches persisted by process_response()
fallback_sweep	Fallback-sweep matches persisted directly (the live parallel-processor path's default)
cli_session	Tags extracted from CLI session files by cli_scanner.py

The CLI path never runs the sweep, so it never emits fallback_sweep/telegram_*; the Telegram path never emits cli_session. Each path owns disjoint source labels, which keeps the two from double-processing the same content.

Next-Session Questions

The <ask-next-session> tag creates questions injected into the next CLI session to maintain conversational continuity (per the session-start protocol in CLAUDE.md). Managed by next_session.py.

Lifecycle:

Question created (pending)
       |
       +---> Injected into CLI session (asked)
       |         via get_injection_text()  (side-effect: marks asked)
       |
       +---> 12h with no CLI session (FALLBACK_DELAY_HOURS)
       |         schedule_fallback_delivery() enqueues a Telegram
       |         prompt via enqueue_prompt()  (dedup_key per day)
       |
       +---> 48h elapsed (DEFAULT_EXPIRY_HOURS)
                 expire_stale() marks expired

Question priority: auto-detected via keyword matching (_detect_priority()). If any word in the question or its context appears in HIGH_PRIORITY_KEYWORDS, priority is high, else normal. The exact constant set:

HIGH_PRIORITY_KEYWORDS = {
    "health", "doctor", "appointment", "pain", "injury", "hurt",
    "commitment", "promise", "deadline", "urgent", "scan", "test",
    "results", "lab", "blood", "surgery", "medication", "dexa",
    "meeting", "reschedule", "cancel", "confirm",
}

These are generic trigger strings, not assertions about anyone — they bias the queue toward health-, commitment-, and time-sensitive follow-ups.

Dedup: new questions are checked against existing pending questions via Jaccard word similarity (_word_overlap()); > WORD_OVERLAP_THRESHOLD (0.8) means duplicate and the add is skipped.

Injection: get_injection_text() returns a formatted block sorted by priority then age, and as a side effect marks the returned questions asked. get_pending_questions() filters out expired/non-pending entries.

Storage: data/stream-captures/next-session-questions.json

{
  "questions": [
    {
      "id": "uuid",
      "content": "How did the <thing> go?",
      "created_at": "2026-01-31T22:43:16-08:00",
      "source_session": "test-pipeline",
      "context": "",
      "priority": "high",
      "expires_at": "2026-02-02T22:43:16-08:00",
      "status": "pending",
      "asked_at": null,
      "fallback_sent": false
    }
  ]
}

Data Structures

Capture Records (JSONL)

File: data/stream-captures/YYYY-MM-DD.jsonl

Each line is a JSON object. (Field set varies slightly by path — the Telegram persister adds content_hash + user_msg; the direct fallback persister adds pattern + confidence; the CLI scanner adds content_hash.) Example (Telegram real-time):

{
  "ts": "2026-02-01T06:43:16.591318+00:00",
  "session": "00000000-0000-0000-0000-000000000000",
  "tag": "file-this",
  "content": "hosted guests for dinner on friday",
  "source": "telegram_realtime",
  "content_hash": "dfbc7cfb7dd29ecc",
  "user_msg": "just got back from hosting dinner"
}

Field	Type	Description
ts	string	ISO8601 timestamp of capture
session	string	Claude session ID (or null)
tag	string	Tag type (one of the five capture tags)
content	string	Extracted content (tag body)
source	string	Processing path that produced this capture
content_hash	string	First 16 chars of SHA-256 of normalized text (Telegram path); full SHA-256 hex (CLI path)
user_msg	string	Truncated user message that triggered the response (max 120 chars)
pattern / confidence	string / float	Present on fallback-sweep records: which detector matched and its confidence

CLI Scanner State

File: data/stream-captures/.cli-scan-state.json

{
  "last_scan_ts": "2026-02-01T06:43:16.591318+00:00",
  "processed_hashes": ["dfbc7cfb...", "7ebc5194..."]
}

Rolling window of the last 1000 hashes; older entries are pruned on each scan.

Consolidation & Sleep Integration

The CLI scan is one sub-task of the micro-consolidation loop, which also handles capture triage (every 5 min), scratch synthesis, and coaching-map refreshes during waking hours. Routing of the captured content happens later:

• Triage (consolidation, ~5-min cadence) reads new captures and begins routing/observation handling.
• /sleep (nightly) reads the day's captures and the next-session queue during reflection — it consumes data/stream-captures/YYYY-MM-DD.jsonl and data/stream-captures/next-session-questions.json, folds file-this/note-to-self into scratch, evaluates observation candidates, and surfaces follow-up items. /sleep does not run the scanner itself.

Tip: Because the scanner runs every 15 minutes (not only at night), CLI-session captures are usually routed within minutes, not held until the nightly cycle.

Components

File	Purpose
scripts/stream_capture/__init__.py	Package init
scripts/stream_capture/telegram_integration.py	Inline tag extraction (whole-text + streaming), persistence, <ask-next-session> routing
scripts/stream_capture/fallback_sweep.py	13 regex patterns for untagged content + dedup/noise filters
scripts/stream_capture/cli_scanner.py	CLI session-file scanner (runs in the consolidation loop)
scripts/stream_capture/next_session.py	Next-session question queue management
scripts/telegram_parallel/parallel_processor.py	Telegram integration point (step 3 of process_request)
scripts/consolidation/__init__.py	MicroConsolidationLoop — invokes scan_cli_sessions() every 15 min
flows/sleep/	Nightly flow that reads (does not produce) stream captures
data/stream-captures/YYYY-MM-DD.jsonl	Daily capture storage
data/stream-captures/next-session-questions.json	Question queue
data/stream-captures/.cli-scan-state.json	CLI scanner state

Usage in CLAUDE.md

Silicon-Zack is instructed to use these tags naturally in responses:

Response with <file-this>hosted guests for dinner</file-this> inline tags
that get <observation>gets energy from hosting large groups</observation>
extracted before delivery.

Tags are invisible to bio-Zack. The response delivered would be:

Response with inline tags that get extracted before delivery.

The public API for reading the queue:

from scripts.stream_capture.next_session import get_pending_questions, add_question

Known Limitations

1. Path divergence in persistence. The reusable process_response() helper persists inline tags (telegram_realtime); the live parallel-processor wiring strips inline tags from output but persists only the fallback sweep's matches (fallback_sweep). A reimplementer should pick one persistence contract and apply it consistently.

2. Fallback sweep false positives. Pattern matching can over-capture. The meta-noise / tool-narration filters, the 20-char substance check, span suppression, and dedup thresholds mitigate this but aren't perfect — fallback captures carry a confidence score precisely because they're heuristic.

3. Session file size cap. The CLI scanner skips files larger than MAX_SESSION_FILE_BYTES (100 MB) to avoid memory blowups.

4. Next-session injection assumes CLI. Questions are designed for CLI session-start injection. If bio-Zack only uses Telegram for an extended period, the 12h fallback (schedule_fallback_delivery()) delivers them via Telegram instead.

Where to Go Next

• Consolidation — the loop that runs the CLI scanner and capture triage on short intervals.
• Sleep Guardian / /sleep — nightly cycle that reads captures during reflection.
• Telegram — the real-time response pipeline that hosts the inline extraction + fallback sweep.
• Scheduling — enqueue_prompt() and the proactive queue used for next-session fallback delivery.