feat(ai): collect raw agent activity in ai discover --history

Ship full AI-agent session activity to GitGuardian, not just MCP tool calls.
ggshield discovers each agent's on-disk transcripts / databases and sends the
raw records verbatim; GitGuardian scans them and strips secrets server-side
before storing them, so the client stays "dumb" and the data shape never
depends on the ggshield version.

Sources cover Claude Code, Codex, Cursor, Copilot CLI and VSCode (Copilot
Chat). Records are batched (by count and by bytes) and posted to the
agent-activity endpoint; the server deduplicates per record.

Issue: NHI-1628

Co-Authored-By: Paul Beslin <paul.beslin-ext@gitguardian.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: 3 people

changelog.d/20260529_000000_paul.beslin.ext_raw_history.md

@@ -0,0 +1,4 @@
+### Changed
+
+- `ggshield ai discover --history` now also ships raw transcript/database records
+  from supported AI agents to GitGuardian.

changelog.d/20260602_000000_achille.mascia_agent_activity.md

@@ -0,0 +1,6 @@
+### Changed
+
+- `ggshield ai discover --history` now collects raw AI-agent activity records
+  from Claude Code, Codex, Cursor, Copilot CLI and VSCode (Copilot Chat) and
+  ships them verbatim to GitGuardian, which scans the content and strips
+  secrets server-side before storing it.

ggshield/cmd/ai/discover.py

@@ -4,7 +4,7 @@
 """
 
 import json
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 import click
 from pygitguardian.models import AIDiscovery
@@ -15,6 +15,10 @@
 from ggshield.core.client import create_client_from_config
 from ggshield.core.errors import APIKeyCheckError, UnknownInstanceError
 from ggshield.core.text_utils import STYLE, format_text, pluralize
+from ggshield.verticals.ai.agent_activity import (
+    AgentActivityReport,
+    collect_agent_activity,
+)
 from ggshield.verticals.ai.agents import AGENTS
 from ggshield.verticals.ai.discovery import (
     discover_ai_configuration,
@@ -72,11 +76,13 @@ def discover_cmd(
         return
 
     backfill_report = BackfillReport()
+    activity_report: Optional[AgentActivityReport] = None
     try:
         config = submit_ai_discovery(client, config)
         save_discovery_cache(config)
         if scan_history:
             backfill_report = backfill_mcp_history(client, config)
+            activity_report = collect_agent_activity(client)
     except Exception as exc:
         if "missing the following scope:" in str(exc):
             scope = str(exc).split("missing the following scope:")[1].strip()
@@ -86,15 +92,19 @@ def discover_cmd(
         ui.display_warning(f"Could not upload AI discovery to GitGuardian: {reason}")
 
     # Summarize after sending to GIM, so we can benefit from its fixes.
-    summary = _summarize_discovery(config, backfill_report)
+    summary = _summarize_discovery(config, backfill_report, activity_report)
 
     if use_json:
         click.echo(json.dumps(summary, indent=2))
     else:
         print_summary(summary)
 
 
-def _summarize_discovery(config: AIDiscovery, report: BackfillReport) -> Dict[str, Any]:
+def _summarize_discovery(
+    config: AIDiscovery,
+    report: BackfillReport,
+    activity_report: Optional[AgentActivityReport] = None,
+) -> Dict[str, Any]:
     """Summarize what we want to show of the discovery."""
     agent_names = set()
     servers = []
@@ -120,7 +130,7 @@ def _summarize_discovery(config: AIDiscovery, report: BackfillReport) -> Dict[st
             }
         )
     servers = sorted(servers, key=lambda x: x["name"])
-    return {
+    summary: Dict[str, Any] = {
         "agents": [AGENTS[name].display_name for name in agent_names],
         "servers": servers,
         "history": {
@@ -130,6 +140,14 @@ def _summarize_discovery(config: AIDiscovery, report: BackfillReport) -> Dict[st
             "skipped": report.skipped,
         },
     }
+    if activity_report is not None:
+        summary["agent_activity"] = {
+            "parsed": activity_report.parsed,
+            "ingested": activity_report.ingested,
+            "duplicates": activity_report.duplicates,
+            "failed_batches": activity_report.failed_batches,
+        }
+    return summary
 
 
 def print_summary(summary: Dict[str, Any]) -> None:
@@ -187,3 +205,19 @@ def print_summary(summary: Dict[str, Any]) -> None:
             f"({history['duplicates']:,} already known, "
             f"{history.get('skipped', 0):,} skipped)"
         )
+
+    agent_activity = summary.get("agent_activity")
+    # Only surface the agent-activity block when there is something to report;
+    # otherwise every `--history` run prints an all-zeros section.
+    if agent_activity is not None and (
+        agent_activity["parsed"] or agent_activity.get("failed_batches", 0)
+    ):
+        click.echo(f"{format_text('Collecting agent activity…', STYLE['key'])}")
+        click.echo(f"  • Parsed {agent_activity['parsed']:,} activity events")
+        click.echo(
+            f"  • Recorded {agent_activity['ingested']:,} activity events "
+            f"({agent_activity['duplicates']:,} already known)"
+        )
+        if agent_activity.get("failed_batches", 0) > 0:
+            label = format_text("Failed batches:", STYLE["detector_line_start"])
+            click.echo(f"  • {label} {agent_activity['failed_batches']:,}")

ggshield/verticals/ai/agent_activity/__init__.py

@@ -0,0 +1,35 @@
+"""AI-agent activity collection: read transcript files and SQLite databases and
+ship each record **raw** (verbatim) to the GitGuardian API, which scans the
+content and strips secrets server-side before storing it."""
+
+from ggshield.verticals.ai.agent_activity.models import AgentActivityEvent
+from ggshield.verticals.ai.agent_activity.orchestrator import (
+    BATCH_SIZE,
+    AgentActivityBatchResult,
+    AgentActivityReport,
+    collect_agent_activity,
+    send_agent_activity_batch,
+)
+from ggshield.verticals.ai.agent_activity.readers import iter_jsonl, iter_sqlite_rows
+from ggshield.verticals.ai.agent_activity.sources import (
+    ActivitySource,
+    JSONActivitySource,
+    JSONLActivitySource,
+    SQLiteActivitySource,
+)
+
+
+__all__ = [
+    "BATCH_SIZE",
+    "ActivitySource",
+    "JSONActivitySource",
+    "JSONLActivitySource",
+    "AgentActivityBatchResult",
+    "AgentActivityEvent",
+    "AgentActivityReport",
+    "SQLiteActivitySource",
+    "collect_agent_activity",
+    "iter_jsonl",
+    "iter_sqlite_rows",
+    "send_agent_activity_batch",
+]

ggshield/verticals/ai/agent_activity/models.py

@@ -0,0 +1,34 @@
+"""Wire shape for raw history events sent to the GitGuardian API."""
+
+from dataclasses import asdict, dataclass
+from typing import Dict
+
+
+@dataclass(frozen=True)
+class AgentActivityEvent:
+    """One raw record from an agent's transcript or database.
+
+    Fields:
+    - ``agent_name``: short agent identifier matching ``Agent.name`` (e.g. ``"claude-code"``).
+    - ``source_kind``: agent-scoped identifier for the file/table the record came from
+      (e.g. ``"session_transcript"``, ``"composer_bubble"``).
+    - ``source_path``: on-disk path relative to the agent's config dir, with variable
+      parts (session UUIDs, workspace hashes) preserved.
+    - ``record_offset``: stable string identifier of the record within its source file.
+      Line index serialised as a string (``"0"``, ``"1"``, …) for JSONL and JSON files.
+      For SQLite, the value(s) of the declared ``key_columns`` — single column: the
+      column value; multiple columns: a JSON-encoded list. Subclasses may override
+      :meth:`ActivitySource.record_offset` for non-trivial cases.
+    - ``content``: the record serialised as a string. JSONL and JSON files: the raw
+      line or file text verbatim. SQLite rows: the row dict serialised with
+      ``json.dumps`` (default) or a custom per-source filter.
+    """
+
+    agent_name: str
+    source_kind: str
+    source_path: str
+    record_offset: str
+    content: str
+
+    def to_dict(self) -> Dict[str, object]:
+        return asdict(self)

ggshield/verticals/ai/agent_activity/orchestrator.py

@@ -0,0 +1,96 @@
+"""Orchestrate agent-activity collection across agents and ship batches to the API."""
+
+import logging
+from dataclasses import dataclass
+from typing import List
+
+import requests
+from pygitguardian import GGClient
+from pygitguardian.models import Detail
+
+from ggshield.verticals.ai.agent_activity.models import AgentActivityEvent
+
+
+logger = logging.getLogger(__name__)
+
+BATCH_SIZE = 500
+
+# Flush a batch once it reaches BATCH_SIZE events OR this many bytes of content,
+# whichever comes first, so a few large records can't produce a huge request.
+MAX_BATCH_BYTES = 5 * 1024 * 1024
+
+
+@dataclass
+class AgentActivityBatchResult:
+    ingested: int
+    duplicates: int
+    success: bool
+
+
+@dataclass
+class AgentActivityReport:
+    parsed: int = 0
+    ingested: int = 0
+    duplicates: int = 0
+    failed_batches: int = 0
+
+
+def send_agent_activity_batch(
+    client: GGClient, events: List[AgentActivityEvent]
+) -> AgentActivityBatchResult:
+    """Serialise ``events`` and submit them as one batch."""
+    if not events:
+        return AgentActivityBatchResult(ingested=0, duplicates=0, success=True)
+    payload = [e.to_dict() for e in events]
+    response = client.send_agent_activity(payload)
+    if isinstance(response, Detail):
+        logger.warning("agent_activity: API returned an error: %s", response.detail)
+        return AgentActivityBatchResult(ingested=0, duplicates=0, success=False)
+    return AgentActivityBatchResult(
+        ingested=response.ingested,
+        duplicates=response.duplicates,
+        success=True,
+    )
+
+
+def collect_agent_activity(client: GGClient) -> AgentActivityReport:
+    """Walk every supported agent's raw sources and ship records in BATCH_SIZE-event batches."""
+    # Imported lazily: agent modules register agent-activity sources that subclass
+    # ``ActivitySource``, so importing ``AGENTS`` at module load would create a
+    # cycle (agents -> agent_activity -> orchestrator -> agents).
+    from ggshield.verticals.ai.agents import AGENTS
+
+    report = AgentActivityReport()
+    buffer: List[AgentActivityEvent] = []
+    buffer_bytes = 0
+
+    def flush() -> None:
+        nonlocal buffer_bytes
+        if not buffer:
+            return
+        # Records are shipped raw: GitGuardian scans and strips secrets
+        # server-side before storing them, so the client stays "dumb".
+        try:
+            result = send_agent_activity_batch(client, list(buffer))
+        except requests.exceptions.RequestException as exc:
+            logger.warning(
+                "agent_activity: batch of %d events failed: %s", len(buffer), exc
+            )
+            report.failed_batches += 1
+        else:
+            report.ingested += result.ingested
+            report.duplicates += result.duplicates
+            if not result.success:
+                report.failed_batches += 1
+        buffer.clear()
+        buffer_bytes = 0
+
+    for agent in AGENTS.values():
+        for event in agent.iter_agent_activity_events():
+            buffer.append(event)
+            buffer_bytes += len(event.content.encode("utf-8"))
+            report.parsed += 1
+            if len(buffer) >= BATCH_SIZE or buffer_bytes >= MAX_BATCH_BYTES:
+                flush()
+    flush()
+    return report

ggshield/verticals/ai/agent_activity/readers.py

@@ -0,0 +1,60 @@
+"""Generic readers for raw history sources (JSONL files, SQLite tables)."""
+
+import logging
+import sqlite3
+from pathlib import Path
+from typing import Dict, Iterator, Sequence, Union
+
+
+logger = logging.getLogger(__name__)
+
+
+def iter_jsonl(path: Path) -> Iterator[str]:
+    """Yield each non-blank line of a JSONL file as a raw string.
+
+    Returns an empty iterator on I/O failure.
+    """
+    if not path.is_file():
+        return
+    try:
+        with path.open("r", encoding="utf-8", errors="replace") as fp:
+            for line in fp:
+                line = line.rstrip("\n").rstrip("\r")
+                if not line.strip():
+                    continue
+                yield line
+    except OSError:
+        return
+
+
+def iter_sqlite_rows(
+    db_path: Path,
+    query: str,
+    params: Sequence[Union[str, int, float, bytes, None]] = (),
+) -> Iterator[Dict[str, object]]:
+    """Yield each row of ``query`` as a ``{column: value}`` dict.
+
+    Missing DB/error → empty iterator.
+    """
+    if not db_path.is_file():
+        return
+    try:
+        conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+    except sqlite3.Error as exc:
+        logger.warning("iter_sqlite_rows: cannot open %s: %s", db_path, exc)
+        return
+    try:
+        conn.row_factory = sqlite3.Row
+        try:
+            cursor = conn.execute(query, params)
+        except sqlite3.Error as exc:
+            logger.warning("iter_sqlite_rows: query failed on %s: %s", db_path, exc)
+            return
+        try:
+            for row in cursor:
+                yield {key: row[key] for key in row.keys()}
+        except sqlite3.Error as exc:
+            logger.warning("iter_sqlite_rows: read failed on %s: %s", db_path, exc)
+            return
+    finally:
+        conn.close()