feat(ai): collect agent activity (raw lines + client redaction) in ai discover --history

Ship full AI-agent session activity (not just MCP calls) to a raw staging
table, keeping the client "dumb" and storing no detected secrets:

- ggshield ships the agent's RAW transcript lines / DB rows verbatim — the data
  shape never depends on the ggshield version (no per-agent field extraction).
- Before sending, each batch is scanned via the GitGuardian secret-scan API
  (multi_content_scan); detected secret spans are redacted client-side and home
  paths anonymised. Fail-closed: a batch that can't be scanned is dropped, never
  shipped.
- Per-record size cap + byte-batching; re-scan every run (offset-skip is a
  follow-up).
- Review fixes to the framework: GGClient typing + Detail handling in the
  orchestrator; POSIX source paths.
- Bumps pygitguardian to send_agent_activity (GitGuardian/py-gitguardian#175).

Aligns with the design doc's staging → canonical structure; the canonical typed
table + per-agent adapters are a follow-up MR. Builds on #1244 / #1257.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: amascia-gg

changelog.d/20260602_000000_achille.mascia_agent_activity.md

@@ -0,0 +1,7 @@
+### Changed
+
+- `ggshield ai discover --history` now ships per-event AI-agent **usage
+  metadata** (event type, tool name, model, timestamps) from Claude Code,
+  Codex and Cursor — not just MCP tool calls. Only an allow-list of safe
+  fields is sent: prompts, command strings, tool inputs/outputs and file
+  contents never leave the machine.

ggshield/cmd/ai/discover.py

@@ -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,
@@ -23,7 +27,6 @@
 )
 from ggshield.verticals.ai.history import BackfillReport, backfill_mcp_history
 from ggshield.verticals.ai.models import Scope
-from ggshield.verticals.ai.raw_history import RawHistoryReport, collect_raw_history
 
 
 @click.command(name="discover")
@@ -73,13 +76,13 @@ def discover_cmd(
         return
 
     backfill_report = BackfillReport()
-    raw_report: Optional[RawHistoryReport] = None
+    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)
-            raw_report = collect_raw_history(client)
+            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()
@@ -89,7 +92,7 @@ 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, raw_report)
+    summary = _summarize_discovery(config, backfill_report, activity_report)
 
     if use_json:
         click.echo(json.dumps(summary, indent=2))
@@ -100,7 +103,7 @@ def discover_cmd(
 def _summarize_discovery(
     config: AIDiscovery,
     report: BackfillReport,
-    raw_report: Optional[RawHistoryReport] = None,
+    activity_report: Optional[AgentActivityReport] = None,
 ) -> Dict[str, Any]:
     """Summarize what we want to show of the discovery."""
     agent_names = set()
@@ -137,12 +140,13 @@ def _summarize_discovery(
             "skipped": report.skipped,
         },
     }
-    if raw_report is not None:
-        summary["raw_history"] = {
-            "parsed": raw_report.parsed,
-            "ingested": raw_report.ingested,
-            "duplicates": raw_report.duplicates,
-            "failed_batches": raw_report.failed_batches,
+    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,
+            "redaction_failures": activity_report.redaction_failures,
         }
     return summary
 
@@ -203,15 +207,24 @@ def print_summary(summary: Dict[str, Any]) -> None:
             f"{history.get('skipped', 0):,} skipped)"
         )
 
-    raw_history = summary.get("raw_history")
-    if raw_history is not None:
-        click.echo(f"{format_text('Collecting raw agent history…', STYLE['key'])}")
-        click.echo(f"  • Parsed {raw_history['parsed']:,} raw events")
+    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 {raw_history['ingested']:,} raw events "
-            f"({raw_history['duplicates']:,} already known)"
+            f"  • Recorded {agent_activity['ingested']:,} activity events "
+            f"({agent_activity['duplicates']:,} already known)"
         )
-        if raw_history.get("failed_batches", 0) > 0:
-            click.echo(
-                f"  • {format_text('Failed batches:', STYLE['detector_line_start'])} {raw_history['failed_batches']:,}"
+        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']:,}")
+        if agent_activity.get("redaction_failures", 0) > 0:
+            label = format_text(
+                "Batches dropped (redaction unavailable):",
+                STYLE["detector_line_start"],
             )
+            click.echo(f"  • {label} {agent_activity['redaction_failures']:,}")

ggshield/verticals/ai/agent_activity/__init__.py

@@ -0,0 +1,35 @@
+"""AI-agent activity collection: read transcript files and SQLite databases,
+extract allow-listed usage metadata per source (never raw free text), and ship
+it to the GitGuardian API."""
+
+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",
+]

…hield/verticals/ai/raw_history/models.py …ld/verticals/ai/agent_activity/models.pyggshield/verticals/ai/raw_history/models.py renamed to ggshield/verticals/ai/agent_activity/models.py ggshield/verticals/ai/raw_history/models.py renamed to ggshield/verticals/ai/agent_activity/models.py

@@ -5,7 +5,7 @@
 
 
 @dataclass(frozen=True)
-class RawHistoryEvent:
+class AgentActivityEvent:
     """One raw record from an agent's transcript or database.
 
     Fields:
@@ -18,7 +18,7 @@ class RawHistoryEvent:
       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:`HistorySource.record_offset` for non-trivial cases.
+      :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.

ggshield/verticals/ai/agent_activity/orchestrator.py

@@ -0,0 +1,110 @@
+"""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
+from ggshield.verticals.ai.agent_activity.redaction import RedactionError, redact_batch
+
+
+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
+    redaction_failures: 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
+        try:
+            # Fail-closed: scan + redact (and anonymise home paths) before
+            # anything is sent. If the scan fails we drop the batch rather than
+            # ship un-scanned content.
+            redacted = redact_batch(client, list(buffer))
+        except RedactionError as exc:
+            logger.warning(
+                "agent_activity: redaction failed for %d events, dropping batch: %s",
+                len(buffer),
+                exc,
+            )
+            report.redaction_failures += 1
+            report.failed_batches += 1
+        else:
+            try:
+                result = send_agent_activity_batch(client, redacted)
+            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