From ae48247f1605a3e999b7945ced3cab59fc707864 Mon Sep 17 00:00:00 2001 From: Eugen Date: Tue, 9 Jun 2026 09:45:39 +0200 Subject: [PATCH] =?UTF-8?q?feat(purchasing):=20raise=20escalate=20threshol?= =?UTF-8?q?d=20to=20=E2=82=AC150k=20+=20decision=20audit=20trail?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related purchasing-engine changes (one effort): 1. Escalate threshold €50k → €150k. Datacenter hardware runs €30k–€110k a line, so a €50k floor escalated every normal buy to sign-off with NO human-approvable path — the cockpit's "Place approved" had nothing to act on. €150k lets routine buys land in `propose` (a human can approve+place them) while still escalating genuinely exceptional spend. The €25k auto-place cap is unchanged, so large buys still never AUTO-place — they require explicit human approval, just not full escalation. Agent-safety fixtures (A3, A7, over-cap) rescaled above €150k; their invariants (at/above threshold → escalate, never auto-placed) are unchanged. 2. DecisionLog append-only audit trail: every purchasing decision is persisted (best-effort, in a SAVEPOINT so a logging failure never fails the run) with a read-only endpoint over it. Mirrors the AssetEvent spine — immutable rows, one per decision per run, recording actor + tier + rationale. Tests: 310 pass, ruff clean. Migration chains after the onboarding table. Demo-first: ship to demo and verify before any prod promotion. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...a2b3c4d5e6_add_decision_log_audit_table.py | 64 ++++++++++ backend/app/api/v1/agent.py | 109 +++++++++++++++- backend/app/core/config.py | 8 +- backend/app/models/__init__.py | 4 + backend/app/models/decision.py | 61 +++++++++ backend/tests/agent_eval/adversarial.py | 12 +- backend/tests/agent_eval/correctness.py | 2 +- backend/tests/test_decision_log.py | 119 ++++++++++++++++++ backend/tests/test_purchasing_run.py | 5 +- 9 files changed, 370 insertions(+), 14 deletions(-) create mode 100644 backend/alembic/versions/f1a2b3c4d5e6_add_decision_log_audit_table.py create mode 100644 backend/app/models/decision.py create mode 100644 backend/tests/test_decision_log.py diff --git a/backend/alembic/versions/f1a2b3c4d5e6_add_decision_log_audit_table.py b/backend/alembic/versions/f1a2b3c4d5e6_add_decision_log_audit_table.py new file mode 100644 index 0000000..6a6e808 --- /dev/null +++ b/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') diff --git a/backend/app/api/v1/agent.py b/backend/app/api/v1/agent.py index 83ad69d..1ebf70a 100644 --- a/backend/app/api/v1/agent.py +++ b/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 diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 9c1408a..31a2a91 100644 --- a/backend/app/core/config.py +++ b/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 diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 58ec31c..e361ac4 100644 --- a/backend/app/models/__init__.py +++ b/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", diff --git a/backend/app/models/decision.py b/backend/app/models/decision.py new file mode 100644 index 0000000..fc54d8a --- /dev/null +++ b/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)) diff --git a/backend/tests/agent_eval/adversarial.py b/backend/tests/agent_eval/adversarial.py index 71f2f05..32e4bbd 100644 --- a/backend/tests/agent_eval/adversarial.py +++ b/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): diff --git a/backend/tests/agent_eval/correctness.py b/backend/tests/agent_eval/correctness.py index bde0f7c..f6298cd 100644 --- a/backend/tests/agent_eval/correctness.py +++ b/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 diff --git a/backend/tests/test_decision_log.py b/backend/tests/test_decision_log.py new file mode 100644 index 0000000..eb6d712 --- /dev/null +++ b/backend/tests/test_decision_log.py @@ -0,0 +1,119 @@ +"""DecisionLog audit-trail tests — the additive decision-loop surfacing. + +Proves, against the real API route (so the best-effort persist in the handler +runs): a purchasing-run writes one append-only DecisionLog row per decision; +GET /agent/decisions lists them (filterable); GET /agent/decisions/{id} drills +one; a placed buy carries its placed_po_id (the provenance join). Copilot is +mocked — no LLM, no network. +""" +from __future__ import annotations + +from datetime import date, timedelta + +from app.agent import copilot +from app.agent.schemas import SourcingRecommendation +from app.models.auth import Role +from app.models.decision import DecisionLog +from app.models.flow import Asset, AssetStatus + +B = "/api/v1" + + +def _mock_copilot(monkeypatch, *, decision="act", confidence=0.95): + def fake(db, product_id, desired_qty=None): + return SourcingRecommendation( + product_id=product_id, recommended_source_id="x", + recommended_qty=desired_qty or 1, rationale="mock", + signals={}, assumptions=[], uncertainties=[], + confidence=confidence, decision=decision) + monkeypatch.setattr(copilot, "recommend_sourcing", fake) + + +def _scenario(client, db_session): + smci = client.post(f"{B}/organizations", json={ + "code": "SMCI", "name": "Supermicro", "is_supplier": True}).json() + srv = client.post(f"{B}/products", json={ + "product_code": "SRV-1U", "name": "1U Server", "category": "server"}).json() + client.post(f"{B}/product-suppliers", json={ + "product_id": srv["id"], "supplier_id": smci["id"], + "standard_lead_time_days": 21, "min_order_quantity": 1, + "contract_price": "3200.00", "preference_rank": 1}) + when = date.today() - timedelta(days=1) + for i in range(3): + db_session.add(Asset(serial_number=f"SN-DL-{i}", product_id=srv["id"], + status=AssetStatus.DECOMMISSIONED, decommissioned_date=when)) + db_session.commit() + return srv + + +def test_purchasing_run_persists_decision_and_lists_it(client, db_session, monkeypatch): + _mock_copilot(monkeypatch) + proc = client.as_role(Role.PROCUREMENT) + srv = _scenario(proc, db_session) + + # Run the gate through the ROUTE (best-effort logging lives in the handler). + res = proc.post(f"{B}/agent/purchasing-run", json={"dry_run": True}).json() + assert any(d["product_id"] == srv["id"] for d in res["decisions"]) + + # It was persisted to the append-only DecisionLog. + rows = db_session.query(DecisionLog).filter(DecisionLog.product_id == srv["id"]).all() + assert rows, "the run's decision must be logged" + assert rows[0].dry_run is True + assert rows[0].actor # the running user's email was captured + + # GET /agent/decisions lists it, newest-first. + listed = proc.get(f"{B}/agent/decisions").json() + assert any(r["product_id"] == srv["id"] for r in listed) + + # Filterable by tier. + tier = rows[0].tier + filtered = proc.get(f"{B}/agent/decisions", params={"tier": tier}).json() + assert filtered and all(r["tier"] == tier for r in filtered) + + # GET /agent/decisions/{id} drills one; unknown id -> 404. + one = proc.get(f"{B}/agent/decisions/{rows[0].id}") + assert one.status_code == 200 and one.json()["id"] == rows[0].id + assert proc.get(f"{B}/agent/decisions/does-not-exist").status_code == 404 + + +def test_placed_decision_links_its_po(client, db_session, monkeypatch): + _mock_copilot(monkeypatch, decision="act", confidence=0.95) + proc = client.as_role(Role.PROCUREMENT) + _scenario(proc, db_session) + + # dry_run=False so an act-tier buy actually places a PO. + res = proc.post(f"{B}/agent/purchasing-run", json={"dry_run": False}).json() + placed = [d for d in res["decisions"] if d.get("placed_po_id")] + if placed: # only assert the join when something actually placed + po_id = placed[0]["placed_po_id"] + row = (db_session.query(DecisionLog) + .filter(DecisionLog.placed_po_id == po_id).first()) + assert row is not None, "a placed decision must log its PO id (provenance join)" + assert row.dry_run is False + + +def test_decisions_endpoint_is_role_gated(client): + # VIEWER cannot read the procurement audit trail. + assert client.as_role(Role.VIEWER).get(f"{B}/agent/decisions").status_code == 403 + # Anonymous is rejected too. + assert client.anon().get(f"{B}/agent/decisions").status_code == 401 + + +def test_logging_failure_never_raises(db_session): + """Best-effort contract: the audit write must swallow any error, never raise. + + Feed _log_decisions a result whose .decisions access throws; the guard has to + absorb it so a logging bug can never fail the purchasing run. + """ + import app.api.v1.agent as agent_mod + + class _Poisoned: + run_at = None + dry_run = True + + @property + def decisions(self): + raise RuntimeError("simulated audit-write failure") + + # Must not raise. + agent_mod._log_decisions(db_session, _Poisoned(), actor="t@e.com") diff --git a/backend/tests/test_purchasing_run.py b/backend/tests/test_purchasing_run.py index fe661cd..a670b83 100644 --- a/backend/tests/test_purchasing_run.py +++ b/backend/tests/test_purchasing_run.py @@ -169,8 +169,9 @@ def test_over_cap_bundle_escalates(client, db_session, monkeypatch): _mock_copilot(monkeypatch, decision="act", confidence=0.99) smci = _org(client, "SMCI", "Supermicro") srv = _product(client, "SRV-1U", "1U Server") - # huge unit price so the bundle clears the escalate threshold (default 50k) - _source(client, srv["id"], smci["id"], price="60000.00") + # huge unit price so the bundle clears the escalate threshold (default 150k): + # 2 * 100k = 200k. + _source(client, srv["id"], smci["id"], price="100000.00") _decommission_assets(db_session, srv["id"], 2) res = purchasing.run_weekly_purchasing(db_session, dry_run=True, period_days=7) dec = next(d for d in res.decisions if d.product_id == srv["id"])