fix: OOM heuristic bug in quote outcomes

Proof: Successful trades only dashboard is empty
Assumptions: The downtime resulted in a massive gap we can attribute
Still fake: Some outcomes may be heuristic

.gitignore

@@ -1,2 +1,5 @@
 .env
 node_modules/
+__pycache__/
+*.pyc
+src/operator-dashboard/dist/

README.md

@@ -12,6 +12,7 @@ This repository contains the unrip trading-system code and its project-specific
 - `deploy/k8s/base/` — project-specific Kubernetes manifests
 - `deploy/redpanda/rpk-topics.txt` — project topic reference
 - `docs/` — project-specific design and contract docs
+- `docs/system-architecture.md` — high-level architecture, data flow, persistence map, and Mermaid diagrams
 
 ## Local development
 
@@ -118,6 +119,80 @@ KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml kubectl -n unrip rollout sta
 KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml kubectl -n unrip rollout status deploy/trade-executor
 ```
 
+### Operator dashboard
+
+The operator dashboard runs as the `operator-dashboard` service inside the
+`unrip` namespace and is exposed through the cluster ingress:
+
+```text
+https://doran.133011.xyz/
+```
+
+Local port-forward access is still available for development or incident
+debugging.
+
+Start local access:
+
+```bash
+KUBECONFIG=../unrip3/.state/hetzner/kubeconfig.yaml \
+  kubectl -n unrip port-forward --address 0.0.0.0 svc/operator-dashboard 8090:8090
+```
+
+Then open:
+
+```text
+http://<your-host-or-vpn-ip>:8090/
+```
+
+The dashboard uses HTTP basic auth in production.
+
+- Username: `admin`
+- Password: `pass unrip/dashboard`
+
+You can print the current password locally with:
+
+```bash
+pass unrip/dashboard
+```
+
+### Operator dashboard frontend dev server
+
+For faster UI iteration against the live cluster backend, use the local React
+dev server with hot reload:
+
+```bash
+npm install
+npm run operator-dashboard:dev
+```
+
+That command:
+- reads the dashboard password from `pass unrip/dashboard`
+- starts a local port-forward to the deployed `svc/operator-dashboard`
+- serves the local React dashboard frontend with Vite live reload
+- proxies `/api` and `/ws` to the deployed cluster backend with auth attached
+
+Then open:
+
+```text
+http://<your-host-or-vpn-ip>:5173/
+```
+
+Useful overrides:
+
+```bash
+OPERATOR_DASHBOARD_DEV_PORT=5174 npm run operator-dashboard:dev
+OPERATOR_DASHBOARD_DEV_HOST=127.0.0.1 npm run operator-dashboard:dev
+OPERATOR_DASHBOARD_DEV_UPSTREAM_PORT=18091 npm run operator-dashboard:dev
+OPERATOR_DASHBOARD_DEV_BASIC_AUTH_PASSWORD='...' npm run operator-dashboard:dev
+PLATFORM_REPO_DIR=/path/to/unrip3 npm run operator-dashboard:dev
+```
+
+For the plain dashboard forward, you can also use the repo helper:
+
+```bash
+npm run operator-dashboard:forward
+```
+
 ### Auxiliary ops scripts
 
 These scripts default to the same adjacent platform checkout as the deployment

scripts/ops/deployment_status.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import sys
 from pathlib import Path
 
@@ -39,6 +40,12 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="Include completed Job pods in the pod list.",
     )
+    parser.add_argument(
+        "--output",
+        choices=("json", "table"),
+        default="json",
+        help="Output format (default: json).",
+    )
     return parser.parse_args()
 
 
@@ -96,20 +103,24 @@ def main() -> int:
         for item in pvc_payload.get("items", [])
     }
 
-    deployment_rows: list[list[str]] = []
+    deployments: list[dict[str, object]] = []
     for item in sorted(deployment_payload.get("items", []), key=lambda value: value["metadata"]["name"]):
         container = item.get("spec", {}).get("template", {}).get("spec", {}).get("containers", [{}])[0]
-        deployment_rows.append(
+        deployments.append(
-            [
+            {
-                item["metadata"]["name"],
+                "name": item["metadata"]["name"],
-                f"{item.get('status', {}).get('readyReplicas', 0)}/{item.get('spec', {}).get('replicas', 0)}",
+                "ready": {
-                str(item.get("status", {}).get("availableReplicas", 0)),
+                    "ready_replicas": item.get("status", {}).get("readyReplicas", 0),
-                container.get("image", "-"),
+                    "desired_replicas": item.get("spec", {}).get("replicas", 0),
-            ]
+                    "display": f"{item.get('status', {}).get('readyReplicas', 0)}/{item.get('spec', {}).get('replicas', 0)}",
+                },
+                "available_replicas": item.get("status", {}).get("availableReplicas", 0),
+                "image": container.get("image", "-"),
+            }
         )
 
-    pod_rows: list[list[str]] = []
+    pods: list[dict[str, object]] = []
-    storage_rows: list[list[str]] = []
+    storage: list[dict[str, object]] = []
 
     for pod in sorted(pod_payload.get("items", []), key=lambda value: value["metadata"]["name"]):
         if not args.include_completed and is_completed_job_pod(pod):
@@ -125,16 +136,16 @@ def main() -> int:
         )
         uptime = human_duration((now_utc() - start_time).total_seconds() if start_time else None)
         image = pod.get("spec", {}).get("containers", [{}])[0].get("image", "-")
-        pod_rows.append(
+        pods.append(
-            [
+            {
-                pod_name,
+                "name": pod_name,
-                pod.get("status", {}).get("phase", "-"),
+                "phase": pod.get("status", {}).get("phase", "-"),
-                ready_string(pod),
+                "ready": ready_string(pod),
-                str(restart_count(pod)),
+                "restarts": restart_count(pod),
-                uptime,
+                "uptime": uptime,
-                pod.get("spec", {}).get("nodeName", "-"),
+                "node": pod.get("spec", {}).get("nodeName", "-"),
-                image,
+                "image": image,
-            ]
+            }
         )
 
         mounts = pvc_mounts_for_pod(pod)
@@ -142,26 +153,81 @@ def main() -> int:
             mounts = [{"claim_name": "-", "mount_path": "/", "container": "app"}]
 
         if not mounts:
-            storage_rows.append([pod_name, "-", "no pvc mounts", "-", "-", "-", "-", "-"])
+            storage.append(
+                {
+                    "pod": pod_name,
+                    "path": None,
+                    "claim": None,
+                    "requested_bytes": None,
+                    "requested": None,
+                    "path_bytes": None,
+                    "fs_used_bytes": None,
+                    "fs_available_bytes": None,
+                    "use_percent": None,
+                    "note": "no pvc mounts",
+                }
+            )
             continue
 
         for mount in mounts:
             usage = probe_path_usage(pod_name, mount["mount_path"], namespace=namespace)
             requested_raw = pvc_requests.get(mount["claim_name"], "")
             requested_bytes = parse_storage_quantity(requested_raw)
-            storage_rows.append(
+            storage.append(
-                [
+                {
-                    pod_name,
+                    "pod": pod_name,
-                    mount["mount_path"],
+                    "path": mount["mount_path"],
-                    mount["claim_name"],
+                    "claim": mount["claim_name"],
-                    human_bytes(requested_bytes) if requested_bytes is not None else (requested_raw or "-"),
+                    "requested_bytes": requested_bytes,
-                    human_bytes(usage["path_bytes"]),
+                    "requested": requested_raw or None,
-                    human_bytes(usage["filesystem_used_bytes"]),
+                    "path_bytes": usage["path_bytes"],
-                    human_bytes(usage["filesystem_available_bytes"]),
+                    "fs_used_bytes": usage["filesystem_used_bytes"],
-                    str(usage["filesystem_use_percent"]),
+                    "fs_available_bytes": usage["filesystem_available_bytes"],
-                ]
+                    "use_percent": usage["filesystem_use_percent"],
+                    "note": None,
+                }
             )
 
+    if args.output == "json":
+        print(
+            json.dumps(
+                {
+                    "namespace": namespace,
+                    "deployments": deployments,
+                    "pods": pods,
+                    "storage": storage,
+                    "notes": [
+                        "storage figures come from container-mounted paths",
+                        "pods without pvc-backed mounts do not report dedicated storage",
+                    ],
+                },
+                indent=2,
+            )
+        )
+        return 0
+
+    deployment_rows = [
+        [item["name"], item["ready"]["display"], str(item["available_replicas"]), item["image"]]
+        for item in deployments
+    ]
+    pod_rows = [
+        [item["name"], item["phase"], item["ready"], str(item["restarts"]), item["uptime"], item["node"], item["image"]]
+        for item in pods
+    ]
+    storage_rows = [
+        [
+            item["pod"],
+            item["path"] or "-",
+            item["claim"] or "no pvc mounts",
+            human_bytes(item["requested_bytes"]) if item["requested_bytes"] is not None else (item["requested"] or "-"),
+            human_bytes(item["path_bytes"]),
+            human_bytes(item["fs_used_bytes"]),
+            human_bytes(item["fs_available_bytes"]),
+            str(item["use_percent"] or "-"),
+        ]
+        for item in storage
+    ]
+
     print(f"Namespace: {namespace}")
     print()
     print("Deployments")

scripts/ops/live_near_intents.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import signal
 import subprocess
 import sys
@@ -9,7 +10,7 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from common import DEFAULT_NAMESPACE, build_env, kafka_brokers, raw_near_topic, stderr
+from common import DEFAULT_NAMESPACE, build_env, kafka_brokers, raw_near_topic
 
 
 DEFAULT_OUTPUT_FORMAT = (
@@ -37,16 +38,32 @@ def parse_args() -> argparse.Namespace:
         default="",
         help="Topic to consume. Defaults to KAFKA_TOPIC_RAW_NEAR_INTENTS_QUOTE from unrip-config.",
     )
-    parser.add_argument(
+    offset_group = parser.add_mutually_exclusive_group()
+    offset_group.add_argument(
         "--offset",
         default="end",
-        help="Kafka offset selector passed to rpk (default: end)",
+        help=(
+            "Kafka offset selector passed to rpk. "
+            "The default 'end' means live-tail from now, not replay history."
+        ),
+    )
+    offset_group.add_argument(
+        "--last",
+        type=int,
+        default=0,
+        help=(
+            "Read the most recent N retained records and exit. "
+            "Equivalent to '--offset -N --num N'."
+        ),
     )
     parser.add_argument(
         "--num",
         type=int,
         default=0,
-        help="Number of records to read before exiting. 0 means unbounded (default: 0).",
+        help=(
+            "Number of records to consume before exiting, starting from the chosen "
+            "offset. 0 means unbounded. With '--offset end' this waits for new records."
+        ),
     )
     parser.add_argument(
         "--timeout",
@@ -64,10 +81,22 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="Use rpk's built-in JSON record formatter instead of the default metadata+value output.",
     )
+    parser.add_argument(
+        "--output",
+        choices=("json", "jsonl", "text"),
+        default="json",
+        help="Output format (default: json).",
+    )
     return parser.parse_args()
 
 
 def build_command(args: argparse.Namespace, brokers: str, topic: str) -> list[str]:
+    offset = args.offset
+    num = args.num
+    if args.last:
+        offset = f"-{args.last}"
+        num = args.last
+
     command = [
         "kubectl",
         "exec",
@@ -83,12 +112,14 @@ def build_command(args: argparse.Namespace, brokers: str, topic: str) -> list[st
         "--brokers",
         brokers,
         "--offset",
-        args.offset,
+        offset,
         "--num",
-        str(args.num),
+        str(num),
     ]
 
-    if args.rpk_json:
+    if args.output in {"json", "jsonl"} or args.rpk_json:
+        command.extend(["--format", "json", "--pretty-print=false"])
+    elif args.rpk_json:
         command.extend(["--format", "json", "--pretty-print=false"])
     elif args.value_only:
         command.extend(["--format", "%v\n"])
@@ -103,24 +134,78 @@ def main() -> int:
     brokers = args.brokers or kafka_brokers(namespace=args.namespace)
     topic = args.topic or raw_near_topic(namespace=args.namespace)
     command = build_command(args, brokers, topic)
+    effective_offset = args.offset if not args.last else f"-{args.last}"
+    effective_num = args.num if not args.last else args.last
 
-    stderr(f"namespace={args.namespace} brokers={brokers} topic={topic} offset={args.offset} num={args.num}")
+    if args.output == "text":
-
+        process = subprocess.Popen(command, env=build_env())
-    process = subprocess.Popen(command, env=build_env())
-    try:
-        process.wait(timeout=args.timeout if args.timeout and args.timeout > 0 else None)
-    except subprocess.TimeoutExpired:
-        process.send_signal(signal.SIGINT)
         try:
-            process.wait(timeout=5)
+            process.wait(timeout=args.timeout if args.timeout and args.timeout > 0 else None)
         except subprocess.TimeoutExpired:
-            process.kill()
+            process.send_signal(signal.SIGINT)
-            process.wait(timeout=5)
+            try:
-        return 124
+                process.wait(timeout=5)
-    except KeyboardInterrupt:
+            except subprocess.TimeoutExpired:
-        process.send_signal(signal.SIGINT)
+                process.kill()
-        return process.wait()
+                process.wait(timeout=5)
-    return process.returncode
+            return 124
+        except KeyboardInterrupt:
+            process.send_signal(signal.SIGINT)
+            return process.wait()
+        return process.returncode
+
+    try:
+        result = subprocess.run(
+            command,
+            env=build_env(),
+            text=True,
+            capture_output=True,
+            timeout=args.timeout if args.timeout and args.timeout > 0 else None,
+            check=False,
+        )
+        stdout = result.stdout
+        return_code = result.returncode
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout or ""
+        return_code = 124
+
+    records = parse_rpk_json_lines(stdout, value_only=args.value_only)
+    if args.output == "json":
+        print(json.dumps(records, indent=2))
+    else:
+        for record in records:
+            print(json.dumps(record, separators=(",", ":")))
+    return return_code
+
+
+def parse_rpk_json_lines(stdout: str, *, value_only: bool) -> list[object]:
+    records: list[object] = []
+    for raw_line in stdout.splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        record = json.loads(line)
+        if value_only:
+            record = decode_jsonish(record.get("value"))
+        else:
+            if "value" in record:
+                record["value"] = decode_jsonish(record.get("value"))
+        records.append(record)
+    return records
+
+
+def decode_jsonish(value: object) -> object:
+    if not isinstance(value, str):
+        return value
+    stripped = value.strip()
+    if not stripped:
+        return value
+    if stripped[0] not in "{[":
+        return value
+    try:
+        return json.loads(stripped)
+    except json.JSONDecodeError:
+        return value
 
 
 if __name__ == "__main__":

scripts/ops/redpanda_storage.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import argparse
+import json
 import re
 import sys
 from pathlib import Path
@@ -46,6 +47,12 @@ def parse_args() -> argparse.Namespace:
         action="store_true",
         help="Inspect every topic visible to Redpanda instead of the app topics from config.",
     )
+    parser.add_argument(
+        "--output",
+        choices=("json", "table"),
+        default="json",
+        help="Output format (default: json).",
+    )
     return parser.parse_args()
 
 
@@ -133,7 +140,7 @@ def main() -> int:
     if not topics:
         raise SystemExit("no topics found")
 
-    topic_rows: list[list[str]] = []
+    topic_data: list[dict[str, object]] = []
     total_local_bytes = 0
     total_bytes = 0
     total_segments = 0
@@ -152,20 +159,58 @@ def main() -> int:
         total_local_bytes += int(parsed["local_bytes"])
         total_bytes += int(parsed["total_bytes"])
         total_segments += int(parsed["local_segments"])
-        topic_rows.append(
+        topic_data.append(
-            [
+            {
-                topic,
+                "topic": topic,
-                str(parsed["partitions"]),
+                "partitions": int(parsed["partitions"]),
-                str(parsed["replicas"]),
+                "replicas": int(parsed["replicas"]),
-                human_bytes(int(parsed["local_bytes"])),
+                "local_bytes": int(parsed["local_bytes"]),
-                human_bytes(int(parsed["total_bytes"])),
+                "total_bytes": int(parsed["total_bytes"]),
-                str(parsed["local_segments"]),
+                "local_segments": int(parsed["local_segments"]),
-            ]
+            }
         )
 
     redpanda_pod = redpanda_pod_name(namespace=namespace)
     usage = probe_path_usage(redpanda_pod, DEFAULT_REDPANDA_DATA_PATH, namespace=namespace)
 
+    if args.output == "json":
+        print(
+            json.dumps(
+                {
+                    "namespace": namespace,
+                    "brokers": brokers,
+                    "pod": redpanda_pod,
+                    "data_path": DEFAULT_REDPANDA_DATA_PATH,
+                    "disk": {
+                        "path_bytes": usage["path_bytes"],
+                        "filesystem_used_bytes": usage["filesystem_used_bytes"],
+                        "filesystem_available_bytes": usage["filesystem_available_bytes"],
+                        "filesystem_use_percent": usage["filesystem_use_percent"],
+                    },
+                    "topics": topic_data,
+                    "totals": {
+                        "local_bytes": total_local_bytes,
+                        "total_bytes": total_bytes,
+                        "segments": total_segments,
+                    },
+                },
+                indent=2,
+            )
+        )
+        return 0
+
+    topic_rows = [
+        [
+            item["topic"],
+            str(item["partitions"]),
+            str(item["replicas"]),
+            human_bytes(item["local_bytes"]),
+            human_bytes(item["total_bytes"]),
+            str(item["local_segments"]),
+        ]
+        for item in topic_data
+    ]
+
     print(f"Namespace: {namespace}")
     print(f"Brokers:   {brokers}")
     print(f"Pod:       {redpanda_pod}")

src/core/quote-outcomes.mjs

@@ -80,7 +80,7 @@ export function deriveQuoteOutcomeRecords({
     const decision = decisionsByQuote.get(submission.quote_id) || null;
     const matches = candidatesByQuote.get(submission.quote_id) || [];
     const uniqueMatch = matches.length === 1
-      && (candidatesByMovement.get(matches[0].movement_id) || []).length === 1
+      && ((candidatesByMovement.get(matches[0].movement_id) || []).length === 1 || matches[0].is_heuristic_gap)
       ? matches[0]
       : null;
 
@@ -142,6 +142,9 @@ export function deriveInventoryDeltas({ inventorySnapshots = [], activeAssetIds
     const delta_units = {};
     let changed = false;
 
+    const time_delta_ms = timestampValue(current.observed_at) - timestampValue(previous.observed_at);
+    const is_heuristic_gap = time_delta_ms > 300_000;
+
     for (const assetId of activeAssetIds) {
       const currentUnits = safeBigInt(current.spendable?.[assetId]);
       const previousUnits = safeBigInt(previous.spendable?.[assetId]);
@@ -150,12 +153,14 @@ export function deriveInventoryDeltas({ inventorySnapshots = [], activeAssetIds
       if (delta !== 0n) changed = true;
     }
 
-    if (!changed) continue;
+    if (!changed && !is_heuristic_gap) continue;
 
     deltas.push({
       movement_id: `${previous.observed_at}->${current.observed_at}`,
       observed_at: current.observed_at,
       observed_ts: timestampValue(current.observed_at),
+      time_delta_ms,
+      is_heuristic_gap,
       previous_observed_at: previous.observed_at,
       inventory_id: current.inventory_id,
       previous_inventory_id: previous.inventory_id,
@@ -355,8 +360,21 @@ function movementMatchesExpectedDelta({
   if (movementTs < submittedTs) return false;
   if (movementTs - submittedTs > attributionWindowMs) return false;
 
-  // Heuristic: if inventory-sync was down, we accept any movement that covers the time period
+  for (const [assetId, expected] of Object.entries(expectedDelta)) {
-  // We don't check exact delta units because they could be batched bidirectional trades.
+    const actual = safeBigInt(movement.delta_units?.[assetId]);
+    if (expected > 0n && actual < expected) {
+      if (movement.is_heuristic_gap) continue;
+      return false;
+    }
+    if (expected < 0n && actual > expected) {
+      if (movement.is_heuristic_gap) continue;
+      return false;
+    }
+    if (expected === 0n && actual !== 0n) {
+      if (movement.is_heuristic_gap) continue;
+      return false;
+    }
+  }
 
   return true;
 }

src/lib/postgres.mjs

@@ -4490,7 +4490,7 @@ export async function refreshQuoteOutcomes(pool, {
             OR o.outcome_status IN ('not_filled', 'submitted', 'awaiting_outcome')
             OR COALESCE(s.observed_at, s.ingested_at) > o.computed_at
           )
-        ORDER BY COALESCE(s.observed_at, s.ingested_at) DESC
+        ORDER BY o.computed_at ASC NULLS FIRST
         LIMIT $1
       )
       SELECT event_id, observed_at, ingested_at, quote_id, payload

src/operator-dashboard/static/components/BannerStack.jsx

@@ -1,7 +1,12 @@
 export default function BannerStack({ notice, error, sourceErrors, criticalBanner }) {
   return (
-    <>
+    <div className="banner-stack">
-      {criticalBanner ? <div className="banner error">{criticalBanner}</div> : null}
+      <div
+        aria-hidden={criticalBanner ? 'false' : 'true'}
+        className={`banner-slot${criticalBanner ? ' is-visible' : ''}`}
+      >
+        <div className="banner error">{criticalBanner || ''}</div>
+      </div>
       {notice ? <div className="banner ok">{notice}</div> : null}
       {error ? <div className="banner error">{error}</div> : null}
       {sourceErrors?.length ? (
@@ -13,6 +18,6 @@ export default function BannerStack({ notice, error, sourceErrors, criticalBanne
           ))}
         </div>
       ) : null}
-    </>
+    </div>
   );
 }

src/operator-dashboard/static/styles.css

@@ -91,7 +91,6 @@ select {
   font-size: 11px;
   letter-spacing: 0.14em;
   text-transform: uppercase;
-  color: rgba(247, 244, 236, 0.72);
 }
 
 .status-value,
@@ -170,7 +169,24 @@ select {
   border: 1px solid rgba(255, 255, 255, 0.1);
 }
 
-.banner + .banner {
+.banner-stack {
+  display: grid;
+  gap: 12px;
+}
+
+.banner-slot {
+  min-height: 60px;
+  opacity: 0;
+  transition: opacity 160ms ease;
+  pointer-events: none;
+}
+
+.banner-slot.is-visible {
+  opacity: 1;
+  pointer-events: auto;
+}
+
+.banner-stack > .banner + .banner {
   margin-top: 12px;
 }