Fix claude-code usage-ingestion freeze (Workflow "failed" rows) + add Weekly plan receipt row

apps/agentacct/Sources/agentacct/ReceiptsPane.swift

@@ -502,7 +502,8 @@ struct RecordSummaryStrip: View {
         if let usd = cost.estimatedCostUsd {
             var qualifier = costBasisLabel(cost.costBasis)
             if cost.costComplete == false { qualifier += " · partial" }
-            if let pct = Fmt.planPct(cost.planShare?.pct) { qualifier += " · \(pct) wkly" }
+            // Weekly-plan share lives in its own "Weekly plan" receipt row now;
+            // don't duplicate it (in a second phrasing) on the cost KPI tile.
             costCell = Cell(
                 id: "cost",
                 label: "Est. cost",
@@ -865,6 +866,10 @@ struct RecordDimensionsCard: View {
                              provenance: receipt.dimensions.cost.provenance,
                              gaps: receipt.dimensions.cost.gaps)
                 hairline
+                dimensionRow("Weekly plan", weeklyPlanSummary,
+                             provenance: nil,
+                             gaps: nil)
+                hairline
                 dimensionRow("Checks", evidenceSummary,
                              provenance: receipt.dimensions.evidence.provenance,
                              gaps: receipt.dimensions.evidence.gaps)
@@ -977,25 +982,24 @@ struct RecordDimensionsCard: View {
             tokensLine = "tokens: " + parts.joined(separator: " · ")
         }
         guard let cost = dim.estimatedCostUsd else {
-            // The share is token-derived, not dollar-derived — an unpriced
-            // task with a calibrated share still states it.
-            var absent = "no priced usage"
-            if let share = dim.planShare?.text { absent += " · \(share)" }
+            let absent = "no priced usage"
             guard let tokensLine else { return absent }
             return absent + "\n" + tokensLine
         }
         let display = receiptCostDisplay(cost, complete: dim.costComplete, confidence: dim.costConfidence)
         var line = "\(display) · \(costBasisLabel(dim.costBasis))\((dim.costComplete ?? true) ? "" : " (partial)")"
-        // The task's share of the weekly plan — shown only once calibrated
-        // (the daemon sends null until then; absence stays a named state on
-        // the merged Usage & limits pane, never a number here).
-        if let share = dim.planShare?.text {
-            line += " · \(share)"
-        }
+        // The Task's weekly-plan share has its own "Weekly plan" row below.
         if let tokensLine { line += "\n" + tokensLine }
         return line
     }
 
+    // The Task's share of its client's weekly plan, as its own receipt row:
+    // the calibrated percentage, or a named calibration state — never a
+    // fabricated number (calibrated-or-nothing). Absent payload → "—".
+    private var weeklyPlanSummary: String {
+        receipt.dimensions.cost.planShare?.rowSummary ?? "—"
+    }
+
     private var evidenceSummary: String {
         let dim = receipt.dimensions.evidence
         return receiptCheckSummary(

apps/agentacct/Sources/agentacct/V1Model.swift

@@ -826,6 +826,21 @@ struct ReceiptPlanShare: Decodable {
         guard let formatted = Fmt.planPct(pct) else { return nil }
         return "\(formatted) of weekly plan"
     }
+
+    /// The dedicated "Weekly plan" receipt row. Calibrated → the percentage
+    /// (≈0% when calibrated-but-negligible, never a bare "—"); otherwise a
+    /// named calibration state, never a fabricated number. Mirrors
+    /// receipt.plan_share_headline so every surface reads identically.
+    var rowSummary: String {
+        if calibrationState == "calibrated", let pct {
+            return (Fmt.planPct(pct) ?? "≈0%") + " of weekly plan"
+        }
+        switch calibrationState {
+        case "calibrating": return "calibrating — not enough 7-day history yet"
+        case "never": return "undefined for this client"
+        default: return "—"
+        }
+    }
 }
 
 struct ReceiptCost: Decodable {

src/agentacct/cli.py

@@ -9480,7 +9480,11 @@ def _find_receipt_task(projection: dict[str, Any], task_id: str) -> dict[str, An
 
 
 def _render_receipt_text(receipt: dict[str, Any]) -> None:
-    from .receipt import evidence_coverage_headline, evidence_coverage_ledger
+    from .receipt import (
+        evidence_coverage_headline,
+        evidence_coverage_ledger,
+        plan_share_headline,
+    )
 
     axes = receipt.get("axes", {})
     dims = receipt.get("dimensions", {})
@@ -9575,6 +9579,7 @@ def _render_receipt_text(receipt: dict[str, Any]) -> None:
 
     cost = dims.get("cost", {})
     table.add_row("Cost", _receipt_cost_text(cost), ", ".join(cost.get("provenance") or []))
+    table.add_row("Weekly plan", plan_share_headline(cost.get("plan_share")), "")
 
     evidence_dim = dims.get("evidence", {})
     table.add_row(

src/agentacct/client_usage.py

@@ -2049,8 +2049,18 @@ def _validate_claude_workflow_journal(
                 )
             keys = set(obj)
             row_type = obj.get("type")
+            # The Workflow tool writes one metadata row per agent lifecycle
+            # transition: "started" and "failed" carry {agentId, key, type};
+            # "result" adds the agent's return value. None of them carry token
+            # usage, so every one is safe to ignore. Any other shape still
+            # fails closed below (e.g. a real assistant/usage row that must not
+            # be silently dropped) -- this is the fail-closed guard, not a
+            # blanket "ignore unknown journals".
             if not (
-                (row_type == "started" and keys == {"agentId", "key", "type"})
+                (
+                    row_type in ("started", "failed")
+                    and keys == {"agentId", "key", "type"}
+                )
                 or (
                     row_type == "result"
                     and keys == {"agentId", "key", "result", "type"}

src/agentacct/receipt.py

@@ -532,6 +532,33 @@ def evidence_coverage_ledger(evidence: Mapping[str, Any]) -> str:
     return " · ".join(bits)
 
 
+def plan_share_headline(plan_share: Mapping[str, Any] | None) -> str:
+    """One honest line for a Task's share of its client's weekly plan.
+
+    Calibrated-or-nothing, the same rule every plan surface honors: a real
+    percentage only once the fit is calibrated; otherwise a named calibration
+    state, never a fabricated number. Mirrors the macOS ``ReceiptPlanShare``
+    wording so the receipt reads identically on every surface.
+    """
+
+    share = plan_share or {}
+    pct = share.get("pct")
+    state = share.get("calibration_state")
+    if state == "calibrated" and isinstance(pct, (int, float)) and not isinstance(pct, bool):
+        if pct >= 0.1:
+            shown = f"≈{pct:.1f}%"
+        elif pct > 0:
+            shown = "≈<0.1%"
+        else:
+            shown = "≈0%"
+        return f"{shown} of weekly plan"
+    if state == "calibrating":
+        return "calibrating — not enough 7-day history yet"
+    if state == "never":
+        return "undefined for this client"
+    return "—"
+
+
 # --- Decision axis ------------------------------------------------------------
 
 def _decision_status(

src/agentacct/tui.py

@@ -1232,7 +1232,11 @@ def _render_receipt_markup(receipt: dict) -> str:
         lines.append("Lifecycle          [b magenta]↗ Handed off[/]")
         if handoff.get("statement"):
             lines.append(f"                   [dim]{_escape(str(handoff['statement']))}[/]")
-    from .receipt import evidence_coverage_headline, evidence_coverage_ledger
+    from .receipt import (
+        evidence_coverage_headline,
+        evidence_coverage_ledger,
+        plan_share_headline,
+    )
 
     lines.append(f"Evidence coverage  [b {ecolor}]{_escape(evidence_coverage_headline(evidence))}[/]")
     _ledger = evidence_coverage_ledger(evidence)
@@ -1298,6 +1302,7 @@ def _prov(name: str) -> str:
 
     cost = dims.get("cost") or {}
     lines.append(f"[b]Cost[/]      {_escape(_receipt_summary_cost(cost))}   [dim]\\[{_prov('cost')}][/]")
+    lines.append(f"[b]Weekly plan[/] {_escape(plan_share_headline(cost.get('plan_share')))}")
 
     ev = dims.get("evidence") or {}
     lines.append(

tests/test_client_usage.py

@@ -3091,6 +3091,54 @@ def tracked(path, **kwargs):
     assert result.events[0].source_parse_complete is True
 
 
+def test_claude_workflow_journal_failed_row_is_ignored(tmp_path):
+    # The Workflow tool records a "failed" lifecycle row when a workflow agent
+    # dies; it carries {agentId, key, type} and no token usage, so it must be
+    # ignored exactly like "started"/"result". Regression: a legitimate
+    # "failed" row tripped claude_workflow_journal_schema_drift, which
+    # fail-closed-aborts the whole home and froze all claude-code usage import.
+    claude_home = _make_claude_home(tmp_path)
+    project = claude_home / "projects" / "-tmp-project"
+    journal = (
+        project
+        / "claude-session"
+        / "subagents"
+        / "workflows"
+        / "wf_failed"
+        / "journal.jsonl"
+    )
+    journal.parent.mkdir(parents=True)
+    journal.write_text(
+        "\n".join(
+            json.dumps(row)
+            for row in (
+                {"agentId": "agent-a", "key": "state", "type": "started"},
+                {
+                    "agentId": "agent-a",
+                    "key": "state",
+                    "result": "ok",
+                    "type": "result",
+                },
+                {"agentId": "agent-b", "key": "state", "type": "failed"},
+            )
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+    result = discover_client_usage_with_diagnostics(
+        client="claude-code",
+        claude_home=claude_home,
+        limit_sessions=10,
+    )
+
+    assert [event.client_session_id for event in result.events] == ["claude-session"]
+    diagnostic = result.diagnostics["claude-code"]
+    assert diagnostic["ignored_non_transcript_files"] == 1
+    assert diagnostic["error_count"] == 0
+    assert diagnostic["error_codes"] == []
+
+
 def test_claude_workflow_journal_schema_drift_fails_closed(tmp_path):
     claude_home = _make_claude_home(tmp_path)
     project = claude_home / "projects" / "-tmp-project"

tests/test_receipt.py

@@ -14,6 +14,7 @@
     RECEIPT_SCHEMA_VERSION,
     build_attention_reason,
     build_receipt,
+    plan_share_headline,
 )
 
 
@@ -659,3 +660,31 @@ def test_receipt_checks_carry_detail_fields_without_command_text() -> None:
     assert row["command_redacted"] is True
     # The store never records command text; the payload must not invent one.
     assert "command" not in row
+
+
+def test_plan_share_headline_is_calibrated_or_nothing() -> None:
+    # Calibrated: a real percentage, with the <0.1% band and an honest ≈0%.
+    assert (
+        plan_share_headline({"pct": 12.2, "calibration_state": "calibrated"})
+        == "≈12.2% of weekly plan"
+    )
+    assert (
+        plan_share_headline({"pct": 0.05, "calibration_state": "calibrated"})
+        == "≈<0.1% of weekly plan"
+    )
+    assert (
+        plan_share_headline({"pct": 0.0, "calibration_state": "calibrated"})
+        == "≈0% of weekly plan"
+    )
+    # Not calibrated: a NAMED state, never a number — even when a pct is present.
+    assert (
+        plan_share_headline({"pct": 9.9, "calibration_state": "calibrating"})
+        == "calibrating — not enough 7-day history yet"
+    )
+    assert (
+        plan_share_headline({"pct": None, "calibration_state": "never"})
+        == "undefined for this client"
+    )
+    # Absent payload stays a dash, never a fabricated zero.
+    assert plan_share_headline(None) == "—"
+    assert plan_share_headline({}) == "—"

tests/test_receipt_cli.py

@@ -383,7 +383,7 @@ def test_receipt_text_render_is_scannable(tmp_path: Path) -> None:
     )["tasks"][0]["task_id"]
     result = runner.invoke(app, ["receipt", task_id, "--store-dir", str(tmp_path)])
     assert result.exit_code == 0, result.output
-    for marker in ("Work Receipt", "Decision status", "Evidence coverage", "Provenance"):
+    for marker in ("Work Receipt", "Decision status", "Evidence coverage", "Weekly plan", "Provenance"):
         assert marker in result.output
     # The evidence line is a coverage RATIO, not a categorical grade word.
     assert "unchecked" in result.output or "checked" in result.output

tests/test_receipt_tui.py

@@ -103,7 +103,7 @@ def test_render_receipt_markup_contains_axes_and_dimensions() -> None:
         "actions": {"tool_category_counts": {}, "tool_category_total": 0, "touched_files": [], "touched_file_count": 0},
     }
     markup = _render_receipt_markup(build_receipt(task, public_task_id="task_x", title="Add rate limit"))
-    for marker in ("Decision status", "Evidence coverage", "[b]Task[/]", "[b]Cost[/]", "Provenance"):
+    for marker in ("Decision status", "Evidence coverage", "[b]Task[/]", "[b]Cost[/]", "[b]Weekly plan[/]", "Provenance"):
         assert marker in markup
     # The evidence line is a coverage RATIO, not a bare axis phrase. Assert the
     # actual content (one completed no-check step reads "1 unchecked") so a