Harden Forge schema boundaries

CONTRACTS.md

@@ -96,6 +96,7 @@ Projection fields:
 - `workflow_ref`
 - `evidence_ref`
 - `workflow_evidence_snapshot`
+- `subject_ref`
 - `assessment_ref`
 - `policy_decision_ref`
 - `use_approval_ref`

README.md

@@ -89,7 +89,7 @@ You can also set an `organization_name` in `config.yaml` or `config.local.yaml`
 | Command | Description |
 |---------|-------------|
 | `forge log` | Log a new incident with interactive prompts |
-| `forge list` | List incidents with `--project`, `--severity`, `--since`, `--tag`, and `--limit` filters |
+| `forge list` | List incidents with `--project`, `--severity`, `--since`, `--tag`, structured-axis, and `--limit` filters |
 | `forge show <id>` | Show full details of one incident; suffix matches like `forge show 001` work |
 | `forge ref <id>` | Print a Proofhouse `IncidentRef` compatibility projection as JSON |
 | `forge edit <id>` | Open an incident in your editor |
@@ -176,7 +176,7 @@ Each incident is a YAML file in `incidents/YYYY-MM/` with structured fields cove
 - resolution: `root_cause`, `immediate_fix`, `systemic_takeaway`
 - metadata: `tags`, `related_incidents`, `playbook_entry`
 - optional Proofhouse axes: `capability_area`, `lifecycle_stage`, `issue_class`, `workflow_archetype`, `subject_type`, `blocked_use_class`, `observed_state`
-- optional pointer refs: `workflow_ref`, `evidence_ref`, `workflow_evidence_snapshot`, `assessment_ref`, `policy_decision_ref`, `use_approval_ref`, `asset_ref`, `derivation_ref`, `transform_ref`
+- optional pointer refs: `workflow_ref`, `evidence_ref`, `workflow_evidence_snapshot`, `subject_ref`, `assessment_ref`, `policy_decision_ref`, `use_approval_ref`, `asset_ref`, `derivation_ref`, `transform_ref`
 
 Existing incident YAML remains compatible: all Proofhouse axes and pointer refs are optional. Older files that only contain the original classification/event/resolution/metadata fields still load, list, analyze, and emit a compatibility `IncidentRef`.
 
@@ -185,12 +185,14 @@ Existing incident YAML remains compatible: all Proofhouse axes and pointer refs
 When an incident relates to Proofhouse workflow evidence or Operational Learning, keep Forge records pointer-based:
 
 - use structured axes for document-operations and Operational Learning failure classes
-- use pointer ref fields for `WorkflowRef`, `EvidenceRef` / `WorkflowEvidenceSnapshot`, `AssessmentRef`, `PolicyDecisionRef`, `UseApprovalRef`, and Operational Learning `AssetRef`, `DerivationRef`, or `TransformRef` placeholders
+- use pointer ref fields for `WorkflowRef`, `EvidenceRef` / `WorkflowEvidenceSnapshot`, `SubjectRef`, `AssessmentRef`, `PolicyDecisionRef`, `UseApprovalRef`, and Operational Learning `AssetRef`, `DerivationRef`, or `TransformRef` placeholders
 - use `context` only for short human-readable summaries
 - use `tags` as secondary discovery aids, not as the only structure
 - use `related_incidents` only for Forge incident IDs
 - do not paste raw customer data, regulated personal data, credentials, or training/eval source material into an incident
 
+Pointer refs and `observed_state` are summary/ref-only fields. Forge rejects obvious raw or sensitive payload keys such as `payload`, `raw_payload`, `document_text`, `claim_text`, `phi`, `ssn`, and `dob`; this is boundary hygiene, not a substitute for upstream redaction.
+
 Governance remains the approval and export-control plane. Forge may record that a handoff or approval issue occurred, but the authoritative rights, redaction, use-approval, manifest, and export state lives outside Forge.
 
 Forge emits a Proofhouse V0.1 `IncidentRef` projection through `forge ref <id>` and the `forge_incident_ref` MCP tool. This projection is generated from saved YAML fields when structured fields exist, and falls back to compatibility inference for older incidents. Until Forge stores structured tenant metadata, the projection uses `organization_id: "unscoped"` and `environment_id: "default"` to avoid implying that `project` is a tenant boundary.

forge_cli/cli.py

@@ -24,6 +24,8 @@
     print_success,
 )
 from forge_cli.incident_store import (
+    AmbiguousIncidentLookupError,
+    DuplicateIncidentError,
     find_incident,
     find_incident_path,
     generate_id,
@@ -38,6 +40,7 @@
     LIFECYCLE_STAGE_VALUES,
     PROOFHOUSE_REF_FIELDS,
     USE_CLASS_VALUES,
+    WORKFLOW_ARCHETYPE_VALUES,
     FailureType,
     Incident,
     Severity,
@@ -186,6 +189,11 @@ def log(
         "--workflow-evidence-snapshot",
         help="WorkflowEvidenceSnapshot pointer as JSON object or snapshot id",
     ),
+    subject_ref: Optional[str] = typer.Option(
+        None,
+        "--subject-ref",
+        help="SubjectRef pointer as JSON object or ref id",
+    ),
     assessment_ref: Optional[str] = typer.Option(
         None,
         "--assessment-ref",
@@ -267,6 +275,9 @@ def log(
         "lifecycle_stage", lifecycle_stage, LIFECYCLE_STAGE_VALUES
     )
     issue_class = _validate_optional_choice("issue_class", issue_class, ISSUE_CLASS_VALUES)
+    workflow_archetype = _validate_optional_choice(
+        "workflow_archetype", workflow_archetype, WORKFLOW_ARCHETYPE_VALUES
+    )
     blocked_use_class = _validate_optional_choice(
         "blocked_use_class", blocked_use_class, USE_CLASS_VALUES
     )
@@ -291,6 +302,7 @@ def log(
             "workflow_ref": workflow_ref,
             "evidence_ref": evidence_ref,
             "workflow_evidence_snapshot": workflow_evidence_snapshot,
+            "subject_ref": subject_ref,
             "assessment_ref": assessment_ref,
             "policy_decision_ref": policy_decision_ref,
             "use_approval_ref": use_approval_ref,
@@ -341,7 +353,11 @@ def log(
         print_info("Incident discarded.")
         raise typer.Exit(0)
 
-    filepath = save_incident(incident, cfg.incidents_dir)
+    try:
+        filepath = save_incident(incident, cfg.incidents_dir)
+    except DuplicateIncidentError as e:
+        print_error(str(e))
+        raise typer.Exit(1)
     print_success(f"Saved: {filepath}")
 
 
@@ -351,6 +367,19 @@ def list_cmd(
     severity: Optional[str] = typer.Option(None, "--severity", "-s", help="Filter by severity"),
     since: Optional[str] = typer.Option(None, "--since", help="Filter by date (YYYY-MM-DD)"),
     tag: Optional[str] = typer.Option(None, "--tag", "-T", help="Filter by tag"),
+    issue_class: Optional[str] = typer.Option(None, "--issue-class", help="Filter by issue class"),
+    capability_area: Optional[str] = typer.Option(
+        None, "--capability-area", help="Filter by capability area"
+    ),
+    lifecycle_stage: Optional[str] = typer.Option(
+        None, "--lifecycle-stage", help="Filter by lifecycle stage"
+    ),
+    workflow_archetype: Optional[str] = typer.Option(
+        None, "--workflow-archetype", help="Filter by workflow archetype"
+    ),
+    blocked_use_class: Optional[str] = typer.Option(
+        None, "--blocked-use-class", help="Filter by blocked use class"
+    ),
     limit: int = typer.Option(10, "--limit", "-n", help="Max incidents to show"),
 ) -> None:
     """List recent incidents with optional filters."""
@@ -366,6 +395,11 @@ def list_cmd(
         severity=severity,
         since=since,
         tag=tag,
+        issue_class=issue_class,
+        capability_area=capability_area,
+        lifecycle_stage=lifecycle_stage,
+        workflow_archetype=workflow_archetype,
+        blocked_use_class=blocked_use_class,
         limit=limit,
     )
 
@@ -383,7 +417,11 @@ def show(
         print_error(str(e))
         raise typer.Exit(1)
 
-    incident = find_incident(cfg.incidents_dir, incident_id)
+    try:
+        incident = find_incident(cfg.incidents_dir, incident_id)
+    except AmbiguousIncidentLookupError as e:
+        print_error(str(e))
+        raise typer.Exit(1)
     if incident is None:
         print_error(f"No incident found matching '{incident_id}'.")
         raise typer.Exit(1)
@@ -403,7 +441,11 @@ def ref_cmd(
         print_error(str(e))
         raise typer.Exit(1)
 
-    incident = find_incident(cfg.incidents_dir, incident_id)
+    try:
+        incident = find_incident(cfg.incidents_dir, incident_id)
+    except AmbiguousIncidentLookupError as e:
+        print_error(str(e))
+        raise typer.Exit(1)
     if incident is None:
         print_error(f"No incident found matching '{incident_id}'.")
         raise typer.Exit(1)
@@ -423,7 +465,11 @@ def edit(
         print_error(str(e))
         raise typer.Exit(1)
 
-    path = find_incident_path(cfg.incidents_dir, incident_id)
+    try:
+        path = find_incident_path(cfg.incidents_dir, incident_id)
+    except AmbiguousIncidentLookupError as e:
+        print_error(str(e))
+        raise typer.Exit(1)
     if path is None:
         print_error(f"No incident found matching '{incident_id}'.")
         raise typer.Exit(1)
@@ -549,6 +595,19 @@ def stats(
     project: Optional[str] = typer.Option(None, "--project", "-p", help="Filter by project"),
     severity: Optional[str] = typer.Option(None, "--severity", "-s", help="Filter by severity"),
     since: Optional[str] = typer.Option(None, "--since", help="Filter by date (YYYY-MM-DD)"),
+    issue_class: Optional[str] = typer.Option(None, "--issue-class", help="Filter by issue class"),
+    capability_area: Optional[str] = typer.Option(
+        None, "--capability-area", help="Filter by capability area"
+    ),
+    lifecycle_stage: Optional[str] = typer.Option(
+        None, "--lifecycle-stage", help="Filter by lifecycle stage"
+    ),
+    workflow_archetype: Optional[str] = typer.Option(
+        None, "--workflow-archetype", help="Filter by workflow archetype"
+    ),
+    blocked_use_class: Optional[str] = typer.Option(
+        None, "--blocked-use-class", help="Filter by blocked use class"
+    ),
 ) -> None:
     """Show summary statistics across all incidents."""
     try:
@@ -565,6 +624,16 @@ def stats(
         incidents = [i for i in incidents if i.severity == severity]
     if since:
         incidents = [i for i in incidents if i.timestamp >= since]
+    if issue_class:
+        incidents = [i for i in incidents if i.issue_class == issue_class]
+    if capability_area:
+        incidents = [i for i in incidents if i.capability_area == capability_area]
+    if lifecycle_stage:
+        incidents = [i for i in incidents if i.lifecycle_stage == lifecycle_stage]
+    if workflow_archetype:
+        incidents = [i for i in incidents if i.workflow_archetype == workflow_archetype]
+    if blocked_use_class:
+        incidents = [i for i in incidents if i.blocked_use_class == blocked_use_class]
 
     display_stats(incidents)
 

forge_cli/display.py

@@ -8,7 +8,7 @@
 from rich.table import Table
 from rich.text import Text
 
-from forge_cli.models import Incident
+from forge_cli.models import PROOFHOUSE_REF_FIELDS, Incident
 
 console = Console()
 
@@ -140,21 +140,7 @@ def display_incident_detail(incident: Incident) -> None:
         lines.append("[cyan bold]Observed state:[/]")
         for key, value in incident.observed_state.items():
             lines.append(f"  {key}: {value}")
-    present_refs = [
-        field_name
-        for field_name in [
-            "workflow_ref",
-            "evidence_ref",
-            "workflow_evidence_snapshot",
-            "assessment_ref",
-            "policy_decision_ref",
-            "use_approval_ref",
-            "asset_ref",
-            "derivation_ref",
-            "transform_ref",
-        ]
-        if getattr(incident, field_name)
-    ]
+    present_refs = [field_name for field_name in PROOFHOUSE_REF_FIELDS if getattr(incident, field_name)]
     if present_refs:
         lines.append("")
         lines.append(f"[cyan bold]Pointer refs:[/]    {', '.join(present_refs)}")
@@ -201,6 +187,8 @@ def display_stats(incidents: list[Incident]) -> None:
     by_project = Counter(i.project for i in incidents)
     by_platform = Counter(i.platform for i in incidents if i.platform)
     all_tags = Counter(tag for i in incidents for tag in i.tags)
+    by_issue_class = Counter(i.issue_class for i in incidents if i.issue_class)
+    by_capability_area = Counter(i.capability_area for i in incidents if i.capability_area)
 
     # Date range
     timestamps = sorted(i.timestamp for i in incidents if i.timestamp)
@@ -235,6 +223,12 @@ def display_stats(incidents: list[Incident]) -> None:
 
     console.print(Columns([type_table, plat_table], equal=True))
 
+    if by_issue_class or by_capability_area:
+        console.print()
+        issue_table = _counter_table("By Issue Class", by_issue_class, color="yellow")
+        capability_table = _counter_table("By Capability Area", by_capability_area, color="cyan")
+        console.print(Columns([issue_table, capability_table], equal=True))
+
     # Top tags
     if all_tags:
         console.print()

forge_cli/incident_store.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 from datetime import date, datetime
+import os
 from pathlib import Path
+import tempfile
 
 import yaml
 
@@ -33,6 +35,14 @@ def _ordered_dict_representer(dumper: yaml.Dumper, data: dict) -> yaml.MappingNo
 # --- ID generation ---
 
 
+class DuplicateIncidentError(FileExistsError):
+    """Raised when saving would overwrite an existing incident file."""
+
+
+class AmbiguousIncidentLookupError(LookupError):
+    """Raised when a suffix lookup matches multiple incidents."""
+
+
 def generate_id(incidents_dir: Path, incident_date: date | None = None) -> str:
     """Generate the next incident ID for the given date (YYYY-MM-DD-NNN)."""
     if incident_date is None:
@@ -63,8 +73,29 @@ def save_incident(incident: Incident, incidents_dir: Path) -> Path:
 
     data = incident.to_dict()
 
-    with open(filepath, "w") as f:
-        yaml.dump(data, f, Dumper=_BlockDumper, default_flow_style=False, allow_unicode=True)
+    if filepath.exists():
+        raise DuplicateIncidentError(f"Incident id already exists: {incident.id}")
+
+    tmp_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            "w",
+            dir=month_dir,
+            prefix=f".{incident.id}.",
+            suffix=".tmp",
+            delete=False,
+        ) as f:
+            tmp_path = Path(f.name)
+            yaml.dump(data, f, Dumper=_BlockDumper, default_flow_style=False, allow_unicode=True)
+            f.flush()
+            os.fsync(f.fileno())
+        try:
+            os.link(tmp_path, filepath)
+        except FileExistsError as exc:
+            raise DuplicateIncidentError(f"Incident id already exists: {incident.id}") from exc
+    finally:
+        if tmp_path is not None and tmp_path.exists():
+            tmp_path.unlink()
 
     return filepath
 
@@ -82,6 +113,11 @@ def list_incidents(
     severity: str | None = None,
     since: str | None = None,
     tag: str | None = None,
+    issue_class: str | None = None,
+    capability_area: str | None = None,
+    lifecycle_stage: str | None = None,
+    workflow_archetype: str | None = None,
+    blocked_use_class: str | None = None,
     limit: int = 10,
 ) -> list[Incident]:
     """List incidents with optional filtering, most recent first."""
@@ -103,6 +139,16 @@ def list_incidents(
             continue
         if tag and tag not in incident.tags:
             continue
+        if issue_class and incident.issue_class != issue_class:
+            continue
+        if capability_area and incident.capability_area != capability_area:
+            continue
+        if lifecycle_stage and incident.lifecycle_stage != lifecycle_stage:
+            continue
+        if workflow_archetype and incident.workflow_archetype != workflow_archetype:
+            continue
+        if blocked_use_class and incident.blocked_use_class != blocked_use_class:
+            continue
 
         incidents.append(incident)
         if len(incidents) >= limit:
@@ -120,9 +166,12 @@ def find_incident_path(incidents_dir: Path, incident_id: str) -> Path | None:
         if exact_path.exists():
             return exact_path
 
-    for filepath in incidents_dir.rglob("*.yml"):
-        if incident_id in filepath.stem:
-            return filepath
+    matches = sorted(filepath for filepath in incidents_dir.rglob("*.yml") if filepath.stem.endswith(incident_id))
+    if len(matches) == 1:
+        return matches[0]
+    if len(matches) > 1:
+        ids = ", ".join(match.stem for match in matches)
+        raise AmbiguousIncidentLookupError(f"Ambiguous incident id '{incident_id}'. Matches: {ids}")
 
     return None