feat(claims): bulk clear for auto-approved claims (#433)

schemas/claim.schema.json

@@ -89,6 +89,11 @@
       "default": null,
       "title": "Approved By"
     },
+    "auto_approved": {
+      "default": false,
+      "title": "Auto Approved",
+      "type": "boolean"
+    },
     "confidence": {
       "default": 0.7,
       "maximum": 1.0,
@@ -141,6 +146,18 @@
       "default": null,
       "title": "Last Confirmed At"
     },
+    "proposed_by": {
+      "anyOf": [
+        {
+          "type": "string"
+        },
+        {
+          "type": "null"
+        }
+      ],
+      "default": null,
+      "title": "Proposed By"
+    },
     "scope": {
       "$ref": "#/$defs/ArtifactScope"
     },

schemas/proposal.schema.json

@@ -5,7 +5,8 @@
         "claim",
         "page",
         "entity",
-        "relation"
+        "relation",
+        "delete"
       ],
       "title": "ProposalKind",
       "type": "string"

src/vouch/capabilities.py

@@ -63,6 +63,7 @@
     "kb.contradict",
     "kb.archive",
     "kb.confirm",
+    "kb.clear_claims",
     "kb.cite",
     "kb.source_verify",
     "kb.session_start",

src/vouch/cli.py

@@ -22,7 +22,7 @@
 import click
 import yaml
 
-from . import __version__, bundle, health, session_split, volunteer_context
+from . import __version__, bundle, health, volunteer_context
 from . import audit as audit_mod
 from . import capture as capture_mod
 from . import codex_rollout as codex_rollout_mod
@@ -64,7 +64,6 @@
     check_approvable,
     expire_pending,
     propose_claim,
-    propose_delete,
     propose_entity,
     propose_page,
     propose_relation,
@@ -1890,28 +1889,67 @@ def archive(claim_id: str) -> None:
     click.echo(f"archived {claim_id}")
 
 
-@cli.command(name="propose-delete")
-@click.argument(
-    "target_kind",
-    type=click.Choice(["claim", "page", "entity", "relation"]),
-)
-@click.argument("target_id")
-@click.option("--rationale", default=None, help="why this should be deleted")
-def propose_delete_cmd(
-    target_kind: str, target_id: str, rationale: str | None
-) -> None:
-    """File a review-gated hard-delete request for an artifact.
+@cli.command(name="claims-clear")
+@click.option("--auto-only", is_flag=True, default=True, show_default=True,
+              help="Clear only auto-approved claims (default: yes)")
+@click.option("--before", type=str, default=None,
+              help="Clear only claims created before this date (ISO 8601, e.g. 2026-07-01)")
+@click.option("--confirm", is_flag=True, default=False,
+              help="Skip confirmation prompt")
+@click.option("--dry-run", is_flag=True, default=False,
+              help="Preview what would be cleared without making changes")
+def claims_clear(auto_only: bool, before: str | None, confirm: bool, dry_run: bool) -> None:
+    """Clear auto-saved claims. Archived claims are preserved in history."""
+    from datetime import datetime
 
-    A different reviewer approves it with `vouch approve <id>`. Refused if the
-    target is still referenced (supersede the claim instead, usually).
-    """
     store = _load_store()
+    before_dt = None
+    if before:
+        try:
+            before_dt = datetime.fromisoformat(before)
+        except ValueError as err:
+            raise click.ClickException(
+                f"invalid date format: {before} (use ISO 8601, e.g. 2026-07-01)"
+            ) from err
+
+    with _cli_errors():
+        to_clear = life.clear_claims(
+            store,
+            auto_only=auto_only,
+            before=before_dt,
+            actor=_whoami(),
+            dry_run=True,  # Always dry-run first to show what will be cleared
+        )
+
+    if not to_clear:
+        click.echo("no claims match the criteria")
+        return
+
+    click.echo(f"found {len(to_clear)} claims to clear:")
+    for claim in to_clear[:10]:  # Show first 10
+        click.echo(f"  {claim.id}: {claim.text[:60]}")
+    if len(to_clear) > 10:
+        click.echo(f"  ... and {len(to_clear) - 10} more")
+
+    if dry_run:
+        click.echo("(dry-run mode: no changes made)")
+        return
+
+    if not confirm and not click.confirm(f"\nClear {len(to_clear)} claims?"):
+        click.echo("cancelled")
+        return
+
+    # Now actually clear them
     with _cli_errors():
-        pr = propose_delete(
-            store, target_kind=target_kind, target_id=target_id,
-            proposed_by=_whoami(), rationale=rationale,
+        life.clear_claims(
+            store,
+            auto_only=auto_only,
+            before=before_dt,
+            actor=_whoami(),
+            dry_run=False,
         )
-    click.echo(f"filed delete proposal {pr.id} for {target_kind} {target_id}")
+
+    click.echo(f"cleared {len(to_clear)} claims")
 
 
 @cli.command()
@@ -1993,13 +2031,6 @@ def session_end_cmd(session_id: str, note: str | None) -> None:
     _emit_json({"session": sess.id, "proposals": sess.proposal_ids})
 
 
-@session.command("list")
-def session_list_cmd() -> None:
-    """List captured sessions in the review pipeline (buffers + pending summaries)."""
-    store = _load_store()
-    _emit_json({"sessions": session_split.build_session_rows(store)})
-
-
 @cli.group()
 def capture() -> None:
     """Automatic session capture (driven by claude code hooks)."""
@@ -2050,10 +2081,8 @@ def capture_observe_cmd() -> None:
 
 @capture.command("finalize")
 @click.option("--session-id", default=None, help="Session id (else read from stdin payload).")
-@click.option("--split/--no-split", "force", default=None,
-              help="Force LLM topical split or a single mechanical page (default: size-gated).")
-def capture_finalize_cmd(session_id: str | None, force: bool | None) -> None:
-    """Roll a session buffer into PENDING summary proposal(s) (SessionEnd hook payload on stdin)."""
+def capture_finalize_cmd(session_id: str | None) -> None:
+    """Roll a session buffer into a PENDING summary (SessionEnd hook payload on stdin)."""
     payload: dict[str, Any] = {}
     if not sys.stdin.isatty():
         raw = sys.stdin.read()
@@ -2073,28 +2102,10 @@ def capture_finalize_cmd(session_id: str | None, force: bool | None) -> None:
     cwd = Path(str(payload.get("cwd") or ".")).resolve()
     transcript_raw = payload.get("transcript_path")
     transcript = Path(str(transcript_raw)) if transcript_raw else None
-    mode = "auto" if force is None else ("split" if force else "mechanical")
     result = capture_mod.finalize(
         store, sid, cwd=cwd, project=cwd.name,
         generated_at=datetime.now(UTC).isoformat(),
-        transcript_path=transcript, mode=mode,
-    )
-    _emit_json(result)
-
-
-@capture.command("summarize")
-@click.argument("session_id")
-@click.option("--split/--no-split", "force", default=None,
-              help="Force split or mechanical (default: size-gated auto).")
-def capture_summarize_cmd(session_id: str, force: bool | None) -> None:
-    """Summarize a captured session into PENDING page proposals (size-gated)."""
-    store = _capture_store()
-    if store is None:
-        _emit_json({"error": "no KB found"})
-        return
-    mode = "auto" if force is None else ("split" if force else "mechanical")
-    result = session_split.summarize(
-        store, session_id, mode=mode, generated_at=datetime.now(UTC).isoformat(),
+        transcript_path=transcript,
     )
     _emit_json(result)
 

src/vouch/jsonl_server.py

@@ -517,6 +517,25 @@ def _h_confirm(p: dict) -> dict:
             if c.last_confirmed_at else None}
 
 
+def _h_clear_claims(p: dict) -> dict:
+    from datetime import datetime
+    before_dt = None
+    if p.get("before"):
+        before_dt = datetime.fromisoformat(p["before"])
+    to_clear = life.clear_claims(
+        _store(),
+        auto_only=p.get("auto_only", True),
+        before=before_dt,
+        actor=_agent(),
+        dry_run=p.get("dry_run", False),
+    )
+    return {
+        "count": len(to_clear),
+        "claim_ids": [c.id for c in to_clear],
+        "dry_run": p.get("dry_run", False),
+    }
+
+
 def _h_cite(p: dict) -> list:
     out = []
     for c in life.cite(_store(), p["claim_id"]):
@@ -796,6 +815,7 @@ def _h_propose_theme(p: dict) -> dict:
     "kb.contradict": _h_contradict,
     "kb.archive": _h_archive,
     "kb.confirm": _h_confirm,
+    "kb.clear_claims": _h_clear_claims,
     "kb.cite": _h_cite,
     "kb.source_verify": _h_source_verify,
     "kb.session_start": _h_session_start,

src/vouch/lifecycle.py

@@ -127,6 +127,73 @@ def confirm(store: KBStore, *, claim_id: str, actor: str) -> Claim:
     return claim
 
 
+def clear_claims(
+    store: KBStore,
+    *,
+    auto_only: bool = True,
+    before: datetime | None = None,
+    actor: str,
+    dry_run: bool = False,
+) -> list[Claim]:
+    """Clear auto-approved claims, optionally filtered by date range.
+
+    Filters claims by auto-approval status and/or created_at timestamp,
+    then archives (not deletes) the matching claims. All operations are
+    audited.
+
+    Args:
+        store: Knowledge base store
+        auto_only: If True, only clear auto-approved claims (auto_approved=True).
+                   If False, clear all claims matching date filter.
+        before: If set, only clear claims created before this datetime.
+        actor: Who is performing the operation.
+        dry_run: If True, don't write changes, just return what would be cleared.
+
+    Returns:
+        List of claims that were (or would be) archived.
+    """
+    all_claims = store.list_claims()
+    to_clear: list[Claim] = []
+
+    for claim in all_claims:
+        # Skip if already archived or status doesn't allow clearing
+        if claim.status == ClaimStatus.ARCHIVED:
+            continue
+
+        # Filter by auto-approval status
+        if auto_only and not claim.auto_approved:
+            continue
+
+        # Filter by date range
+        if before and claim.created_at >= before:
+            continue
+
+        to_clear.append(claim)
+
+    # Apply the clear operation (archive the claims)
+    if not dry_run:
+        for claim in to_clear:
+            claim.status = ClaimStatus.ARCHIVED
+            claim.updated_at = datetime.now(UTC)
+            store.update_claim(claim)
+
+        # Log the bulk operation
+        if to_clear:
+            audit.log_event(
+                store.kb_dir,
+                event="claim.bulk_clear",
+                actor=actor,
+                object_ids=[c.id for c in to_clear],
+                data={
+                    "count": len(to_clear),
+                    "auto_only": auto_only,
+                    "before": before.isoformat() if before else None,
+                },
+            )
+
+    return to_clear
+
+
 def cite(store: KBStore, claim_id: str) -> list[Evidence | dict]:
     """Return resolved citations for a claim.
 

src/vouch/models.py

@@ -267,6 +267,8 @@ def _text_non_empty(cls, v: str) -> str:
     updated_at: datetime = Field(default_factory=utcnow)
     last_confirmed_at: datetime | None = None
     approved_by: str | None = None  # vouch: review-gate audit
+    proposed_by: str | None = None  # vouch: tracks who proposed this claim
+    auto_approved: bool = False  # vouch: true if approved by proposer (trusted-agent mode)
 
     @field_validator("scope", mode="before")
     @classmethod

src/vouch/proposals.py

@@ -524,7 +524,13 @@ def approve(
         _ensure_no_existing_artifact(store, proposal.kind, payload["id"])
     result: Claim | Page | Entity | Relation
     if proposal.kind == ProposalKind.CLAIM:
-        claim = Claim(approved_by=approved_by, **payload)
+        is_auto_approved = approved_by == proposal.proposed_by
+        claim = Claim(
+            approved_by=approved_by,
+            proposed_by=proposal.proposed_by,
+            auto_approved=is_auto_approved,
+            **payload
+        )
         store.put_claim(claim)
         with index_db.open_db(store.kb_dir) as conn:
             index_db.index_claim(

src/vouch/server.py

@@ -734,6 +734,42 @@ def kb_confirm(claim_id: str) -> dict[str, Any]:
     return {"id": c.id, "last_confirmed_at": c.last_confirmed_at}
 
 
+@mcp.tool()
+def kb_clear_claims(
+    auto_only: bool = True,
+    before: str | None = None,
+    dry_run: bool = False,
+) -> dict[str, Any]:
+    """Clear auto-approved claims, optionally filtered by date range.
+
+    Args:
+        auto_only: If True, only clear auto-approved claims.
+        before: If set, only clear claims created before this ISO 8601 date.
+        dry_run: If True, preview without making changes.
+
+    Returns:
+        Dictionary with count of cleared claims and their IDs.
+    """
+    from datetime import datetime
+
+    before_dt = None
+    if before:
+        before_dt = datetime.fromisoformat(before)
+
+    to_clear = life.clear_claims(
+        _store(),
+        auto_only=auto_only,
+        before=before_dt,
+        actor=_agent(),
+        dry_run=dry_run,
+    )
+    return {
+        "count": len(to_clear),
+        "claim_ids": [c.id for c in to_clear],
+        "dry_run": dry_run,
+    }
+
+
 @mcp.tool()
 def kb_cite(claim_id: str) -> list[dict[str, Any]]:
     """Return resolved citations (sources or evidence records) backing a claim."""