feat(purchasing): raise escalate threshold to €150k + decision audit trail

backend/alembic/versions/f1a2b3c4d5e6_add_decision_log_audit_table.py

@@ -0,0 +1,64 @@
+"""add decision_log audit table
+
+Append-only audit trail for the autonomous purchasing decision engine. Additive
+only — CREATE TABLE plus its indexes; no existing table is touched.
+
+Revision ID: f1a2b3c4d5e6
+Revises: db90c46c5938
+Create Date: 2026-06-09 10:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'f1a2b3c4d5e6'
+down_revision: Union[str, Sequence[str], None] = 'db90c46c5938'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        'decision_log',
+        sa.Column('run_at', sa.String(length=32), nullable=False),
+        sa.Column('dry_run', sa.Boolean(), nullable=False),
+        sa.Column('product_id', sa.String(length=36), nullable=False),
+        sa.Column('supplier_id', sa.String(length=36), nullable=True),
+        sa.Column('qty', sa.Integer(), nullable=False),
+        sa.Column('unit_price', sa.Float(), nullable=True),
+        sa.Column('total', sa.Float(), nullable=False),
+        sa.Column('trigger_type', sa.String(length=48), nullable=True),
+        sa.Column('evidence', sa.JSON(), nullable=True),
+        sa.Column('tier', sa.String(length=16), nullable=False),
+        sa.Column('confidence', sa.Float(), nullable=True),
+        sa.Column('rationale', sa.Text(), nullable=True),
+        sa.Column('placed_po_id', sa.String(length=36), nullable=True),
+        sa.Column('actor', sa.String(length=128), nullable=True),
+        sa.Column('id', sa.String(length=36), nullable=False),
+        sa.Column('date_created', sa.DateTime(), nullable=False),
+        sa.Column('last_updated', sa.DateTime(), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+    )
+    with op.batch_alter_table('decision_log', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_decision_log_run_at'), ['run_at'], unique=False)
+        batch_op.create_index(batch_op.f('ix_decision_log_product_id'), ['product_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_decision_log_supplier_id'), ['supplier_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_decision_log_tier'), ['tier'], unique=False)
+        batch_op.create_index(batch_op.f('ix_decision_log_placed_po_id'), ['placed_po_id'], unique=False)
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    with op.batch_alter_table('decision_log', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_decision_log_placed_po_id'))
+        batch_op.drop_index(batch_op.f('ix_decision_log_tier'))
+        batch_op.drop_index(batch_op.f('ix_decision_log_supplier_id'))
+        batch_op.drop_index(batch_op.f('ix_decision_log_product_id'))
+        batch_op.drop_index(batch_op.f('ix_decision_log_run_at'))
+
+    op.drop_table('decision_log')

backend/app/api/v1/agent.py

@@ -7,6 +7,7 @@
 """
 from __future__ import annotations
 
+from datetime import datetime
 from typing import List, Optional
 from uuid import UUID
 
@@ -19,6 +20,8 @@
 from app.agent.schemas import AgentInsight, PurchasingRunResult, SourcingRecommendation
 from app.api.deps import get_current_user, get_db, require_role
 from app.models.auth import Role, User
+from app.models.decision import DecisionLog
+from app.services.exceptions import NotFoundError
 
 router = APIRouter(tags=["agent"], prefix="/agent")
 
@@ -141,20 +144,118 @@ def ask(payload: AskRequest, db: Session = Depends(get_db),
         raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc))
 
 
+def _log_decisions(db: Session, result: PurchasingRunResult, actor: Optional[str]) -> None:
+    """Persist each decision to the append-only DecisionLog — BEST EFFORT.
+
+    A logging failure must never fail the run, so every write happens inside a
+    SAVEPOINT (begin_nested) and the whole block is guarded: if it raises, we
+    roll back only the log writes and swallow the error, leaving any placed POs
+    and the run result untouched. Insert-only — rows are never updated.
+    """
+    try:
+        for d in result.decisions:
+            try:
+                with db.begin_nested():
+                    db.add(DecisionLog(
+                        run_at=result.run_at.isoformat(),
+                        dry_run=result.dry_run,
+                        product_id=d.product_id,
+                        supplier_id=d.supplier_id,
+                        qty=d.qty,
+                        unit_price=d.unit_price,
+                        total=d.total,
+                        trigger_type=getattr(d.trigger, "type", None),
+                        evidence=getattr(d.trigger, "evidence", None),
+                        tier=d.tier,
+                        confidence=d.confidence,
+                        rationale=d.rationale,
+                        placed_po_id=d.placed_po_id,
+                        actor=actor,
+                    ))
+            except Exception:  # noqa: BLE001 — best-effort per-row; skip the bad one
+                continue
+    except Exception:  # noqa: BLE001 — logging must never break the run
+        pass
+
+
 @router.post("/purchasing-run", response_model=PurchasingRunResult,
              dependencies=[Depends(_purchasing_role)])
-def purchasing_run(payload: PurchasingRunRequest, db: Session = Depends(get_db)):
+def purchasing_run(payload: PurchasingRunRequest, db: Session = Depends(get_db),
+                   user: User = Depends(get_current_user)):
     """Run the weekly purchasing automation. dry_run=True (default) places nothing."""
-    return purchasing.run_weekly_purchasing(
+    result = purchasing.run_weekly_purchasing(
         db, dry_run=payload.dry_run, period_days=payload.period_days)
+    _log_decisions(db, result, actor=user.email)   # best-effort audit write
+    return result
 
 
 @router.post("/purchasing-run/confirm", response_model=PurchasingRunResult,
              dependencies=[Depends(_purchasing_role)])
-def purchasing_run_confirm(payload: PurchasingConfirmRequest, db: Session = Depends(get_db)):
+def purchasing_run_confirm(payload: PurchasingConfirmRequest, db: Session = Depends(get_db),
+                           user: User = Depends(get_current_user)):
     """Approve->place: recompute the run and place POs only for approved suppliers
     whose recomputed bundle is placeable (act/propose). Escalate bundles are never
     placed here. Returns the recomputed run with placed_po_id set on confirmed ones."""
-    return purchasing.run_weekly_purchasing(
+    result = purchasing.run_weekly_purchasing(
         db, period_days=payload.period_days,
         approve_suppliers=set(payload.approve_suppliers))
+    _log_decisions(db, result, actor=user.email)   # best-effort audit write
+    return result
+
+
+# --- decision audit trail (read-only over the append-only DecisionLog) --------
+
+class DecisionLogOut(BaseModel):
+    id: str
+    run_at: str
+    dry_run: bool
+    product_id: str
+    supplier_id: Optional[str] = None
+    qty: int
+    unit_price: Optional[float] = None
+    total: float
+    trigger_type: Optional[str] = None
+    evidence: Optional[dict] = None
+    tier: str
+    confidence: Optional[float] = None
+    rationale: Optional[str] = None
+    placed_po_id: Optional[str] = None
+    actor: Optional[str] = None
+    date_created: datetime
+
+    model_config = {"from_attributes": True}
+
+
+@router.get("/decisions", response_model=List[DecisionLogOut],
+            dependencies=[Depends(_purchasing_role)])
+def list_decisions(db: Session = Depends(get_db),
+                   tier: Optional[str] = None,
+                   product_id: Optional[str] = None,
+                   supplier_id: Optional[str] = None,
+                   run_at: Optional[str] = None,
+                   placed_only: bool = False,
+                   limit: int = 200):
+    """The persistent decision audit trail, newest first. All filters optional."""
+    q = db.query(DecisionLog)
+    if tier:
+        q = q.filter(DecisionLog.tier == tier)
+    if product_id:
+        q = q.filter(DecisionLog.product_id == product_id)
+    if supplier_id:
+        q = q.filter(DecisionLog.supplier_id == supplier_id)
+    if run_at:
+        q = q.filter(DecisionLog.run_at == run_at)
+    if placed_only:
+        q = q.filter(DecisionLog.placed_po_id.isnot(None))
+    q = q.order_by(DecisionLog.date_created.desc()).limit(max(1, min(limit, 1000)))
+    return list(q)
+
+
+@router.get("/decisions/{decision_id}", response_model=DecisionLogOut,
+            dependencies=[Depends(_purchasing_role)])
+def get_decision(decision_id: str, db: Session = Depends(get_db)):
+    """One decision by id (drill into its inputs; placed_po_id joins provenance)."""
+    row = db.get(DecisionLog, decision_id)
+    if row is None:
+        raise NotFoundError(f"Decision {decision_id!r} not found")
+    return row

backend/app/core/config.py

@@ -51,7 +51,13 @@ def _normalize_db_url(cls, v: str) -> str:
     # Weekly purchasing automation — gates and defaults (all env-overridable).
     auto_place_spend_cap: float = 25000.0      # ACT bundles above this can't auto-place
     act_confidence_floor: float = 0.8          # min copilot confidence to auto-place
-    escalate_spend_threshold: float = 50000.0  # bundle total at/above this -> escalate
+    # Calibrated for datacenter hardware: a single server/GPU line routinely runs
+    # €30k–€110k, so a €50k escalate floor sent every normal buy to sign-off with
+    # no human-approvable path. €150k lets routine buys land in `propose` (a human
+    # can approve+place them in the cockpit) while still escalating genuinely
+    # exceptional spend. The auto-place cap stays €25k, so large buys still never
+    # AUTO-place — they require an explicit human approval, just not full escalation.
+    escalate_spend_threshold: float = 150000.0  # bundle total at/above this -> escalate
     replace_ratio: float = 1.0                 # replacements per decommissioned unit
     default_reorder_floor: int = 0             # per-product floor when none is set
 

backend/app/models/__init__.py

@@ -21,6 +21,9 @@
     CostParams,
     ShouldCostRun,
 )
+from app.models.decision import (  # noqa: F401
+    DecisionLog,
+)
 from app.models.flow import (  # noqa: F401
     Asset,
     AssetEvent,
@@ -92,6 +95,7 @@
     "AssetStatus",
     "AssetEvent",
     "AssetEventType",
+    "DecisionLog",
     "TrkSupplier",
     "TrkPurchaseOrder",
     "Shipment",

backend/app/models/decision.py

@@ -0,0 +1,61 @@
+"""Append-only audit trail for the autonomous purchasing decision engine.
+
+This mirrors :class:`app.models.flow.AssetEvent` — it is the *trail* of what the
+agent decided and why, one immutable row per decision per run. Nothing here is
+ever updated or deleted in normal operation: like the asset event spine, it is
+written once (INSERT only) and only ever read back, so a run is fully
+reconstructable after the fact (who ran it, what each decision was, which tier
+it landed in, and — for placed buys — which PurchaseOrder it became, so the
+existing provenance chain attaches through ``placed_po_id``).
+
+It is deliberately additive: the decision engine
+(:func:`app.agent.purchasing.run_weekly_purchasing`) is untouched. The route
+handler persists each decision here as a *best-effort* write, so a logging
+failure can never fail the run itself.
+"""
+from __future__ import annotations
+
+from typing import Optional
+
+from sqlalchemy import JSON, Float, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.db import Base, IdMixin, TimestampMixin
+
+
+class DecisionLog(IdMixin, TimestampMixin, Base):
+    """One persisted purchasing decision from a purchasing-run.
+
+    Append-only: rows are inserted by the purchasing-run route and never mutated.
+    ``date_created`` (from :class:`TimestampMixin`) is the decision timestamp.
+    """
+
+    __tablename__ = "decision_log"
+
+    # Which run this decision belongs to (the run's run_at, ISO string) + whether
+    # that run was a dry run. Lets the audit table group a preview's decisions.
+    run_at: Mapped[str] = mapped_column(String(32), index=True)
+    dry_run: Mapped[bool] = mapped_column(default=True)
+
+    # The decision itself (mirrors agent.schemas.PurchasingDecision).
+    product_id: Mapped[str] = mapped_column(String(36), index=True)
+    supplier_id: Mapped[Optional[str]] = mapped_column(String(36), index=True)
+    qty: Mapped[int] = mapped_column(Integer)
+    unit_price: Mapped[Optional[float]] = mapped_column(Float)
+    total: Mapped[float] = mapped_column(Float, default=0.0)
+
+    # Why it was justified + how it was judged.
+    trigger_type: Mapped[Optional[str]] = mapped_column(String(48))
+    evidence: Mapped[Optional[dict]] = mapped_column(JSON)   # the numbers behind the trigger
+    tier: Mapped[str] = mapped_column(String(16), index=True)  # act | propose | escalate
+    confidence: Mapped[Optional[float]] = mapped_column(Float)
+    rationale: Mapped[Optional[str]] = mapped_column(Text)
+
+    # Set only when the buy was actually placed — the join into the existing
+    # provenance chain (PurchaseOrder -> OrderItem -> Asset). Free text FK to
+    # purchase_order.id; kept un-constrained so a logging write never depends on
+    # the PO row being flushed/visible yet (best-effort by design).
+    placed_po_id: Mapped[Optional[str]] = mapped_column(String(36), index=True)
+
+    # Who triggered the run (free text, same convention as AssetEvent.actor).
+    actor: Mapped[Optional[str]] = mapped_column(String(128))

backend/tests/agent_eval/adversarial.py

@@ -84,11 +84,11 @@ def _a2_expect(result, world: World, db):
 def _a3_setup(db) -> World:
     sup = make_supplier(db, "Pricey Co")
     prod = make_product(db, "ADV-OVERCAP", category="server")
-    # 10 units * 5_000 = 50_000 -> at/above escalate_spend_threshold (50k).
-    make_source(db, prod, sup, contract_price=5000.0, moq=1)
+    # 10 units * 20_000 = 200_000 -> at/above escalate_spend_threshold (150k).
+    make_source(db, prod, sup, contract_price=20000.0, moq=1)
     decommission(db, prod, 10)
     return World(product_id=prod.id, supplier_id=sup.id,
-                 extra={"total": 50_000.0})
+                 extra={"total": 200_000.0})
 
 
 def _a3_expect(result, world: World, db):
@@ -162,12 +162,12 @@ def _a6_expect(result, world: World, db):
 def _a7_setup(db) -> World:
     sup = make_supplier(db, "Stale Co")
     prod = make_product(db, "ADV-STALE", category="server")
-    # On live recompute this bundle is 12 * 5_000 = 60_000 -> escalate-tier, which
+    # On live recompute this bundle is 12 * 15_000 = 180_000 -> escalate-tier, which
     # a confirm/approval can NEVER place. Approving the supplier is the "forged"
     # replay; recompute-from-live overrides it.
-    make_source(db, prod, sup, contract_price=5000.0, moq=1)
+    make_source(db, prod, sup, contract_price=15000.0, moq=1)
     decommission(db, prod, 12)
-    return World(product_id=prod.id, supplier_id=sup.id, extra={"total": 60_000.0})
+    return World(product_id=prod.id, supplier_id=sup.id, extra={"total": 180_000.0})
 
 
 def _a7_expect(result, world: World, db):

backend/tests/agent_eval/correctness.py

@@ -7,7 +7,7 @@
 Settings the assertions key off (app/core/config.py):
   act_confidence_floor      = 0.8       # below -> never 'act'
   auto_place_spend_cap      = 25_000    # 'act' requires total <= this
-  escalate_spend_threshold  = 50_000    # at/above -> 'escalate'
+  escalate_spend_threshold  = 150_000   # at/above -> 'escalate'
 """
 from __future__ import annotations