orionarchitekton/skills/ops/import-conversation-logs

import-conversation-logs

Lane: ops
Version: 0.1.0
Owner: system
Schedule: 23:28 PT daily

Purpose

Ingest conversations from multiple AI chat platforms into the unified messages_v2 table. This is the foundation of the memory pipeline - all other skills depend on this data.

Sources

| Platform | Inbox Path | Format | |----------|-----------|--------| | ChatGPT | /cosmocrat/inbox/raw/chatgpt/ | OpenAI export JSON | | Claude | /cosmocrat/inbox/raw/claude/ | Anthropic export JSON | | Gemini | /cosmocrat/inbox/raw/gemini/ | Google export JSON | | Cursor | /cosmocrat/inbox/raw/cursor/ | SQLite global_state.vscdb | | Codex | /cosmocrat/inbox/raw/codex/ | JSON files | | Grok | /cosmocrat/inbox/raw/grok/ | JSON files |

Memory Contract

Writes

| Type | Table | Retention | Description | |------|-------|-----------|-------------| | conversation_message | messages_v2 | permanent | All chat messages | | conversation_topic | conversation_topics | permanent | Extracted topics |

Access Mode

service_only - All writes go through MemoryClient. No direct ClickHouse access.

PII Handling

• Mode: redact
• Fields: content
• Uses app/tools/pii_utils.py for redaction

Provenance

Every ingested message includes:

| Field | Source | Example | |-------|--------|---------| | run_id | Generated | ingest_2025-12-22_a1b2c3d4 | | trace_id | Langfuse or run_id | trace_xyz789 | | source_ref | Platform:path | chatgpt:/cosmocrat/inbox/raw/chatgpt/web/2025-12-22/export.json | | actor | Fixed | import-conversation-logs | | intent | Optional | daily_conversation_ingest |

Pipeline Position

multi-source-collector (23:22 PT)
         ↓
    inbox files
         ↓
import-conversation-logs (23:28 PT) ← YOU ARE HERE
         ↓
    messages_v2 table
         ↓
conversation-exporter (23:35 PT)
         ↓
executive-daily-report (00:02 PT)

Implementation

File: app/services/mcp/jobs/ingest/daily_ingest_runner.py

Key Functions

• main() - Entry point, orchestrates all sources
• process_inbox_source() - Per-source processing
• check_routing_mismatches() - Signature validation
• archive_processed_files() - Post-processing cleanup

Using MemoryClient

from app.services.memory import MemoryClient, MemoryRecord, Provenance

memory = MemoryClient()
provenance = Provenance(
    run_id=f"ingest_{date}_{uuid4().hex[:8]}",
    trace_id=os.environ.get("LANGFUSE_TRACE_ID", ""),
    source_ref=f"{source_name}:{inbox_path}",
    actor="import-conversation-logs"
)

records = [
    MemoryRecord(
        lane="ops",
        type="conversation_message",
        namespace="cosmocrat",
        content=row,
        metadata={"source": source_name}
    )
    for row in rows
]

results = memory.upsert_mem_batch(records, provenance, pii_mode="redact")

Error Handling

| Error | Action | Alert | |-------|--------|-------| | Empty inbox | Log warning, continue | No | | Source processor error | Log error, continue to next source | Slack summary | | Insert batch failure | Retry once, log partial results | Slack summary | | Routing mismatch | Quarantine file, log warning | Slack if >5 files |

Guards

1. skip_if_empty_inbox - No-op if all inboxes empty
2. validate_source_signatures - Detect misrouted files
3. quarantine_misrouted_files - Move bad files to /cosmocrat/quarantine/

Environment Variables

| Variable | Required | Description | |----------|----------|-------------| | CLICKHOUSE_URL | Yes | ClickHouse HTTP endpoint | | CLICKHOUSE_USER | Yes | ClickHouse username | | CLICKHOUSE_PASSWORD | Yes | ClickHouse password | | INBOX_ROOT | No | Default: /cosmocrat/inbox/raw | | ARCHIVE_ROOT | No | Default: /cosmocrat/conversation_archive | | LANGFUSE_TRACE_ID | No | For observability linking | | COSMO_NOTIFY_SLACK | No | Default: true |

Testing

# Run locally
python -m app.services.mcp.jobs.ingest.daily_ingest_runner

# With backfill
python -m app.services.mcp.jobs.ingest.daily_ingest_runner --backfill-days 7

# Integration tests
pytest tests/skills/test_import_conversation_logs.py -v

Metrics

| Metric | Type | Description | |--------|------|-------------| | ingest.conversation.duration_ms | Gauge | Total run time | | ingest.conversation.messages_count | Counter | Messages ingested | | ingest.conversation.sources_processed | Counter | Sources with data | | ingest.conversation.files_processed | Counter | Files processed | | ingest.conversation.errors_count | Counter | Errors encountered | | ingest.conversation.quarantined_count | Counter | Files quarantined |

Changelog

• 0.1.0 (2025-12-22): Initial manifest, migrated to MemoryClient