Drive PyObs Desk from your own code
The page reads pasted Python in the browser for free — one or more files joined into one
paste, each introduced by a # file: <path> line — with its line numbers. It
lexes strings and comments out first, rebuilds the block structure, inventories every log call,
prometheus_client metric, OpenTelemetry span, except block, outbound HTTP
call and request or task handler, and runs 25 deterministic checks drawn from Python observability
practice (catalogue ids PO-01..PO-25, plus PO-26 for context): print() as the log,
f-string messages, the root logger, swallowed exceptions and lost tracebacks, credentials, personal
data and whole payloads in log calls, metrics created inside a function, invalid metric names,
high-cardinality labels and .labels() calls that raise, Summaries, wall-clock timing,
spans that never end, span names built from data, failures swallowed inside a span, a
TracerProvider without service.name, logs with no trace id, outbound calls that drop the
context, no request id, and handlers nothing measures. Its two paid steps are available over HTTP.
You send one JSON object with task set to "audit" or
"instrument", the paste as code and the page’s check as a
facts string, and the model replies with one JSON object: an audit that rates the four
signals, answers every flag and gives exact before/after edits per finding, or the whole
instrumented file(s) with a list of every signal it added.
Base URL: https://api.skillsafe.ai/v1/app-api. Every call sends
Content-Type: application/json and, once you have a token,
Authorization: Bearer <token>. The paths below (/guest,
/me, /estimate, /run, /jobs/{job_id},
/run-stream) are relative to it, exactly as sdk.js calls them. A
plain-text summary of the app for agents is at /llms.txt. Both lanes run on
the gpt-terra model with a 10% publisher markup.
The response envelope
Responses are JSON of the form {"ok": ..., "data": ..., "error": {...}}. The page’s
SDK (sdk.js) decides success from the HTTP status, not from ok: on a
2xx it returns data; otherwise it raises an error whose message is
error.message (or the HTTP status text when the body carries none), with
status, code (error.code) and details
(error.details) attached. A body that is not JSON is treated as {}.
{ "ok": true, "data": { ... } }
{ "ok": false, "error": { "code": "...", "message": "...", "details": { ... } } }
Error codes
Branch on the HTTP status and error.code, show error.message. The page
handles 401 and 402 by name; the rest reach it as a generic failure (“The run failed: ...”).
| HTTP | error.code | What it means / what to do |
|---|---|---|
400 | VALIDATION_ERROR | The body is not a JSON object, a required field (task, code, facts) is missing, or a declared field has the wrong type. Every field is a string; sending facts or audit_handoff as an object is the usual cause. |
401 | UNAUTHORIZED | No token, or one that expired or was revoked. The page forgets the signed-in user and says “Your session expired - sign in again.” Get a new token. |
402 | INSUFFICIENT_CREDITS | The balance cannot cover the run. The page says “Not enough credits for this run.” with a top-up link, and tries never to submit into a 402: it disables the run button when the /me balance is below the estimate’s min_credits. |
404 | NOT_FOUND | A wrong path, or a job id that does not belong to this token. |
409 | CONFLICT | An Idempotency-Key replayed with a different body. Send the original body, or use a new key (the page bumps its :a<attempt> suffix for a retry). |
429 | RATE_LIMITED | Too many requests. Back off with a delay; do not tight-loop, and never resend a metered run without its original Idempotency-Key. |
Failures that do not arrive as an HTTP status:
| What | What happens |
|---|---|
event: error on /run-stream | The SDK raises an error carrying the event’s message, code and job_id. |
Job status: "failed" | Polling stops; the job’s error field says why. |
| Polling past 180 s | The SDK’s own client-side limit (waitForJob, 1 s interval, 180 000 ms default): it throws “job timed out”. The job itself is not cancelled. |
The task field: two lanes
task routes the run, and it takes exactly two values. Each lane has its own reply
contract, and a reply never mixes them. The skill’s instructions tell the model to do the
audit lane when task is missing or anything else (and the page’s own
Obskit.buildInput turns any lane that is not "instrument" into
"audit"), so a typo silently gets you an audit: always send one of the two values.
task | What it does | verdict | Use it when |
|---|---|---|---|
"audit" | Reviews the service the way an SRE does before it goes to production: says what the code does and emits today, rates the four signals (logs, metrics, traces, correlation) as good, partial or missing, confirms or dismisses every page flag, adds what text rules cannot see, and gives the verbatim evidence and exact before/after edits for each finding, listed in the order to apply them — what breaks or leaks first (metric errors, credentials in logs, swallowed failures, unbounded labels), then missing signals (tracebacks, correlation, latency per handler), then hygiene. | blocked | needs_work | ready | Before shipping: would you see it if it failed, is the telemetry itself safe, and what should change first. |
"instrument" | Writes the instrumented version of the pasted code: the whole of every Python file it changes (and at most one new module), with a latency Histogram labelled by method, route template and status per handler, traffic and errors, spans with constant names around the work and each outbound call, a request id in middleware that every log line carries and outbound calls send, and a traceback on every failure path — plus a signals list of every metric and span, the setup steps for process start, and what it left open. | instrumented | partial | When you want the code changed, not a list of findings: a service with no metrics or traces yet, or what is left after an audit. |
The page chains the two: after an audit, “Apply every edit and instrument the rest” puts
every audit edit into the paste, switches to the instrument lane on that edited paste and
adds an audit_handoff string (below) with the audit’s verdict, headline and
findings. Over the API that is a second run with task: "instrument", the edited paste as
code, facts recomputed from the edited paste, and that string. The page
drops the handoff as soon as the paste is edited again.
Guards the page applies before a run and the API does not: it refuses to price or run a paste with no Python it can read (“No Python file was found. Paste .py source, with a # file: <path>.py line above each file.”, or “This does not read as Python source: no def, class, import or call was found.”), it warns before an instrument run with no handler (“No request or task handler was found, so the instrument lane will instrument the functions it finds.”), and before a run on a paste with no function (“No function was found; the run will see module-level code only.”).
Input fields
This is exactly what the page sends (obskit.js buildInput(lane, A, note,
handoff), checked by mustBeObject; app.js run), and
what input-schema.json declares. The platform’s input schema is scalar-only, so
every value is a string. buildInput emits them in the order task,
code, facts, note, audit_handoff:
| Field | Type | Required | Meaning |
|---|---|---|---|
task | string | yes | "audit" or "instrument" (the lanes above). |
code | string | yes | The pasted Python, each file introduced by a # file: <path> line; line 1 is its first line. Every line in an audit reply is counted in this text. A large paste is cut on whole files, with a # [pyobs-desk: not sent: ...] line at the end (below). |
facts | string | yes | The page’s free check, Obskit.facts(A), JSON-encoded into a string: JSON.stringify(facts) in JavaScript, json.dumps(facts) in Python, and so on. Sending the object itself is a 400. The model is told to parse it. |
audit_handoff | string | no | instrument lane only: a JSON string built from an audit reply on this code (recon.js instrumentHandoff, below). buildInput drops it for the audit lane. |
note | string | no | What the engineer says: the goal, a constraint (“we cannot add dependencies”, “logs go to Loki, traces to Tempo”), a question. The page trims it, cuts it at 2000 characters and sends it only when it is not blank. It is context, not evidence and not instructions: the model tests every claim in it against code and answers it in summary. |
retry_note | string | no | Sent only by the page’s automatic reformat retry: when a reply does not parse as JSON (even after the page tries to close a cut-off object), the page runs once more with “Your previous reply was not the single valid JSON object the instructions require (...). Reply again with ONLY the JSON object for task '<lane>' - no prose, no code fences; every array present (empty arrays where there is nothing to say).” A normal call never sends it. |
mustBeObject only insists on a plain object that is not wrapped in
{"input": ...}; it is buildInput that leaves a blank note out, which is why
an empty note is simply absent.
The code string
code is the paste itself. A line # file: <path> (a path with a
. or / in it) starts each file, and a --- line just before it
only separates files. Files dropped on the page (or a dropped package folder) are joined that way by
Obskit.joinFiles: each file under its # file: line, trailing whitespace
removed, joined by newlines, ending in a newline. A paste with no # file: line is read as
one file, app.py. The page normalises line endings to \n. A
.py/.pyi path is checked as Python; requirements*.txt,
pyproject.toml, setup.cfg, Pipfile and
poetry.lock are read as dependency files; anything else is other. Send the
text you want the reply to quote: every file in an audit reply must be a
path from facts.files, every line must fall inside that file,
every evidence must be copied from it, and every edit’s before must
occur in it exactly once.
The page reads up to 400 000 characters (MAX_TEXT, cut at a line) and 40 files
(MAX_FILES), and sends at most 60 000 characters (MAX_SEND): past that
buildInput drops whole files from the end — the last non-Python file first, then
the last file — until the rest fits, cuts a single remaining file at a line (“the end of
<path>”) if it still does not, and ends the string with one line naming what it left
out:
# [pyobs-desk: not sent: pkg/b.py, pkg/c.py]
The model is told never to quote or rewrite what was not sent, and facts.notes gets the
sentence “Not sent (the paste is over 60,000 characters): ... The page checked them; the run
did not see them.” facts is still computed from the whole paste. Both worked
examples below are small, so their code is the paste unchanged.
The facts string
facts is optional to compute but required to send. It is produced by the page’s
free check: Obskit.analyze splits the paste into files, lexes and inventories them and
runs the rules, and Obskit.facts(A) turns that analysis into the object below (it does
not depend on the lane: an audit and an instrument run of the same paste carry the same facts). It is
how the model knows which flags to answer, which file paths and line ranges to name, and which
handlers and functions must survive. Its top-level keys, in the order built:
| Key | Contents |
|---|---|
service | framework (FastAPI, Flask, Django, Celery, ... or null), libraries (of structlog, loguru, logging, prometheus_client, opentelemetry), python_files and handlers (counts). |
files | Every file: path (use it verbatim), kind (python, deps, other), line and end_line (the paste lines of the file’s content, without its # file: line). |
handlers | Request and task handlers: name, file, line, kind (http, task), method, route. |
functions | Every def (up to 200): name (Class.method for methods), file, line. |
logs | Every log call (up to 150): file, line, level, lib (stdlib, structlog, loguru), style (literal, fstring, percent, format, concat, variable), message (shortened to 70 characters), fields, exc (the traceback is logged). |
metrics | Every prometheus_client metric: var, name, full (with namespace/subsystem), type, labels, file, line, uses (how many places record it). |
spans | Every span: name, api (start_as_current_span, start_span), style, file, line. |
outbound, excepts | Outbound HTTP calls (file, line, call) and every except (up to 120: file, line, type, broad, raises). |
flags | Every flag the page raised: uid (F1, F2, ... in severity, file, then line order), id (PO-01..PO-26), sev (high | medium | low | info), file, line, text, detail, and lines only on a flag that aggregates several lines (every line it covers). An audit must answer every non-info flag by its uid. |
verdict_candidate | The audit verdict the flags point at (the page’s own reading): blocked with any high flag, needs_work with any medium, otherwise ready. |
notes | Plain sentences about what was not read or not sent (empty for a paste that fits). |
The simplest source is obskit.js itself: the page loads it from
/obskit.js, and it also runs in Node (it sets module.exports).
This script prints what the page would send as code and facts (with the
60 000-character cut applied), with the facts decoded so you can read them; the step 4 samples compact
them back into the page’s string byte for byte. Save the paste as
worker-paste.txt and redirect the output into worker-check.json:
// build-check.js -- run: node build-check.js worker-paste.txt > worker-check.json (obskit.js saved next to it)
const fs = require("fs");
const Obskit = require("./obskit.js");
const A = Obskit.analyze(fs.readFileSync(process.argv[2], "utf8"));
if (!A.ok) { console.error(A.errors.join("\n")); process.exit(1); }
const inp = Obskit.buildInput("audit", A, "", null); // code and facts do not depend on the lane
console.log(JSON.stringify({ code: inp.code, facts: JSON.parse(inp.facts) }, null, 2));
const real = A.flags.filter((f) => f.sev !== "info");
console.error(A.files.length + " file(s), " + A.handlers.length + " handler(s); " + real.length + " flags (" +
A.counts.high + " high, " + A.counts.medium + " medium, " + A.counts.low + " low); verdict_candidate " + A.candidate);
For the worker example it reports 1 file(s), 0 handler(s); 7 flags (0 high, 3 medium, 4 low); verdict_candidate needs_work; for the quotes example,
1 file(s), 2 handler(s); 2 flags (0 high, 2 medium, 0 low); verdict_candidate needs_work. For both worked examples its code and
facts are exactly what the page sends.
Without the page’s check. The platform only checks that facts
is a string, so a minimal object works — "{}", or a hand-built one such as
{"flags": [], "notes": ["No page check was run; read the code directly."]}
— and the model still reads code. But you lose what the check adds: with no
flags there is nothing to confirm or dismiss, with no files there are no
paths and line ranges to name, and with no handlers the instrument lane has no list of
handlers it must cover. Build it with obskit.js as above when you can.
The audit_handoff string
Only for task: "instrument", and only for the code the audit was about with every
audit edit applied. The page builds both from the normalised audit reply
res (recon.js normalize, below) and the analysis
A of the audited paste, in recon.js instrumentHandoff:
const run = Recon.applyFindings(A, res.findings, Obskit); // every finding's edits, in order, each on top of the last
JSON.stringify({ verdict: res.verdict, headline: res.headline,
findings: res.findings.map(f => ({ id: f.id, severity: f.severity, file: f.file, line: f.line, title: f.title, why: f.why, edits: f.edits.length })),
applied: /* edits that applied */, failed: /* edits that did not */ })
// run.A.text is the edited paste: it becomes code, and facts is recomputed from it
That is {"verdict", "headline", "findings": [{id, severity, file, line, title, why, edits}],
"applied", "failed"}, where each finding’s edits is the number of
edits it carried. applyFindings applies every finding’s edits with
Obskit.applyEdits, in the order listed: each before must occur exactly once
in its file as the earlier edits left it (leading and trailing blank lines ignored), or that edit is
reported as failed and skipped. It is a previous reply, not evidence: the model checks it against
code and instruments what is left. This script does the whole step — the edited
paste, its recomputed facts and the handoff string, in the same shape as
build-check.js prints, so the samples read it the same way:
// handoff.js -- run: node handoff.js worker-paste.txt worker-reply.txt > worker-instrument-check.json (obskit.js, recon.js next to it)
const fs = require("fs");
const Obskit = require("./obskit.js");
const Recon = require("./recon.js");
const A = Obskit.analyze(fs.readFileSync(process.argv[2], "utf8")); // the audited paste
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), "audit");
const ho = Recon.instrumentHandoff(A, res); // every edit, in order
const B = Obskit.analyze(ho.text); // facts recomputed from the edited paste
const inp = Obskit.buildInput("instrument", B, "", ho.handoff);
console.log(JSON.stringify({ code: inp.code, facts: JSON.parse(inp.facts), audit_handoff: inp.audit_handoff }, null, 2));
console.error(ho.applied + " edit(s) applied, " + ho.failed + " did not apply");
Save the audit’s output.output as worker-reply.txt; for the worker
example it reports 12 edit(s) applied, 0 did not apply.
The output contracts
The model is instructed (SKILL.md) to reply with one JSON object and
nothing else — no prose, no code fences — with every array of its lane
present ([] when empty). The run’s output text is that object. In both lanes it
must ground everything in code: a module, middleware, logging configuration, dashboard,
alert, exporter or collector that is not in the paste is never claimed to exist — it goes into
questions. It must never write a secret value: a line that holds a literal credential is
never copied into the reply (not in evidence, not in an edit, not in a file), and no credential,
real hostname or real endpoint is invented. It must never change what the code does: same functions,
names, arguments, return values, responses and status codes; a new try/except
it adds ends with a bare raise. In the audit lane every page flag whose sev
is not info is answered exactly once in flag_answers, and an info flag is
never reported as a finding.
task: "audit"
task:"audit".verdict:blocked(any finding is high) |needs_work(the worst is medium) |ready(nothing above low; low findings and questions do not stop a release).headline: one sentence.service: one or two sentences — what the code does and what it emits today.signals[]: exactly four entries, in the order logs, metrics, traces, correlation:{signal, status, note}withstatusgood|partial|missing.findings[]:{id, flags[], file, line, severity, pillar, title, evidence, why, edits[]}.idis a catalogue id (PO-01..PO-25) orother;flagslists the pageuids it confirms ([]for something the page did not flag);fileis apathfromfacts.files, copied exactly;lineis the line ofcodewith the offending text, inside that file’sline..end_line;severityishigh(the telemetry itself raises, a span is never exported, a credential is logged, a broad except hides failures completely, or a label is unbounded) |medium(a signal is missing or misleading) |low(hygiene);pillarislogs|metrics|traces|correlation;evidenceis one to three whole lines copied verbatim fromcodeas pasted.edits[]:{before, after}on that one file. Eachbeforeis a contiguous, exact slice of the file — whole lines with their indentation — that occurs exactly once in the file as it stands when the edit is applied;afteris its replacement. Findings are applied in the order listed and their edits in order, each on top of the ones before, so a later finding that touches the same lines writes itsbeforeas the earlier one left it.editsmay be[]only for a low finding or a change outside the paste.flag_answers[]:{flag, status, reason}, exactly one per non-info page flag,statusconfirmed(a finding lists that uid) |dismissed(with the reason).plan[]: the findings in the order to apply them.questions[](only what the paste cannot answer) andsummary(two to four sentences, answeringnoteif one was sent).
task: "instrument"
task:"instrument".verdict:instrumented(openis empty) |partial.headline: one sentence.files[]:{path, content}— every pasted Python file it changes, whole (every line, without the# file:line, four spaces, never tabs), under itspathfromfacts.files; at most one new module (for example<package>/telemetry.py) next to the pasted files. It uses the libraries the code already imports and adds onlyprometheus_clientandopentelemetry-api; a TracerProvider, exporter or logging handlers are configured in a request module only when the paste already does.signals[]:{kind, name, type, labels[], where, why}withkindmetric|span|logandtypehistogram|counter|gauge|span|log. Every metric and span in the returned code is listed with the same name, type and labels, and every entry is in the code; log events are optional.whereis the function (ormodule, or the middleware) that records it.setup[]: what must be configured at process start or in the deployment, one step each (dependencies, logging format, TracerProvider withservice.nameand exporter, the/metricsendpoint, environment variables) — as steps, never as a claim that they exist.open[]:{item, reason}for what it left uninstrumented.questions[]andsummary, as in the audit lane; with anaudit_handoff, it instruments what the audit left.
How the page reads a reply (recon.js parseResult and
normalize): it strips a code fence and parses from the first { to the
last }. A reply is read as the lane that was asked for, unless it names the other lane
and carries that lane’s body (a files array for instrument, a
findings array for audit). An unknown verdict shows as
needs_work or partial and counts as a disagreement; an unknown
severity becomes medium; ids like PO7 or po 7
become PO-07; flags keeps only F<n> values; findings
with no file, evidence and title are dropped; file may also arrive as
path, and an open item as path; questions is
capped at 10, plan at 30 and setup at 20. An audit reply carries 9
sections (verdict, headline, service, signals, findings, flag_answers, plan, questions, summary) and
an instrument reply 8 (verdict, headline, files, signals, setup, open, questions, summary), which is
what the page counts when it recovers a partial reply.
Worked examples
Both are examples the page ships (the example buttons on the app page; the full
pastes are in /example.js, ids worker and
quotes), built exactly as the page builds them:
Obskit.buildInput(lane, Obskit.analyze(text), note), with a real model reply for each.
The pastes, their facts and the audit reply are shown whole; the request bodies and the instrumented
file inside the instrument reply are trimmed for display, and each trim is marked. What is shown is
verbatim. Neither paste holds a credential.
Audit: an invoice worker whose failed renders report success
One file, invoices/worker.py: a queue worker with no web framework that pops billing
jobs off a BillingQueue and renders each invoice, using structlog,
prometheus_client (a Counter and a Summary) and OpenTelemetry (a TracerProvider with an
OTLP exporter and one span per render). The page’s free check raises 7 flags (3 medium, 4 low) (plus the
PO-26 context flag); its own reading of the verdict is needs_work. The
engineer’s note: “Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?”
The code string, whole: 2,018 characters, 59 lines ending in a
newline. Save it as worker-paste.txt and run
node build-check.js worker-paste.txt > worker-check.json:
# file: invoices/worker.py
"""Renders monthly invoices for Quillmoor Energy customers from the billing queue."""
import time
import structlog
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from prometheus_client import Counter, Summary
from invoices.render import render_pdf
from invoices.queue import BillingQueue, JobFailed
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
logger = structlog.get_logger()
INVOICES_RENDERED = Counter("invoices_rendered_total", "Invoices rendered", ["outcome"])
RENDER_TIME = Summary("invoice_render_duration_seconds", "Time to render one invoice")
queue = BillingQueue("invoices")
def render_invoice(job):
with tracer.start_as_current_span(f"render invoice {job.invoice_id}") as span:
span.set_attribute("invoice.id", job.invoice_id)
started = time.time()
try:
pdf = render_pdf(job.invoice_id, job.template)
except JobFailed as err:
logger.warning("invoice render failed", invoice_id=job.invoice_id, error=str(err))
INVOICES_RENDERED.labels(outcome="failed").inc()
return None
RENDER_TIME.observe(time.time() - started)
INVOICES_RENDERED.labels(outcome="ok").inc()
logger.info("invoice rendered", invoice_id=job.invoice_id, pages=pdf.pages)
return pdf
def run_forever():
while True:
job = queue.pop(timeout=30)
if job is None:
continue
try:
render_invoice(job)
queue.ack(job)
except Exception as exc:
logger.error("job crashed", job_id=job.id, error=str(exc))
queue.nack(job)
if __name__ == "__main__":
print("invoice worker starting")
run_forever()
The facts object, decoded and whole. The string the page sends is its compact
JSON.stringify, 4,574 characters:
{
"service": {
"framework": null,
"libraries": [
"structlog",
"prometheus_client",
"opentelemetry"
],
"python_files": 1,
"handlers": 0
},
"files": [
{
"path": "invoices/worker.py",
"kind": "python",
"line": 2,
"end_line": 60
}
],
"handlers": [],
"functions": [
{
"name": "render_invoice",
"file": "invoices/worker.py",
"line": 28
},
{
"name": "run_forever",
"file": "invoices/worker.py",
"line": 44
}
],
"logs": [
{
"file": "invoices/worker.py",
"line": 35,
"level": "warning",
"lib": "structlog",
"style": "literal",
"message": "invoice render failed",
"fields": [
"invoice_id",
"error"
],
"exc": false
},
{
"file": "invoices/worker.py",
"line": 40,
"level": "info",
"lib": "structlog",
"style": "literal",
"message": "invoice rendered",
"fields": [
"invoice_id",
"pages"
],
"exc": false
},
{
"file": "invoices/worker.py",
"line": 53,
"level": "error",
"lib": "structlog",
"style": "literal",
"message": "job crashed",
"fields": [
"job_id",
"error"
],
"exc": false
}
],
"metrics": [
{
"var": "INVOICES_RENDERED",
"name": "invoices_rendered_total",
"full": "invoices_rendered_total",
"type": "Counter",
"labels": [
"outcome"
],
"file": "invoices/worker.py",
"line": 22,
"uses": 2
},
{
"var": "RENDER_TIME",
"name": "invoice_render_duration_seconds",
"full": "invoice_render_duration_seconds",
"type": "Summary",
"labels": [],
"file": "invoices/worker.py",
"line": 23,
"uses": 1
}
],
"spans": [
{
"name": "f\"render invoice {job.invoice_id}\"",
"api": "start_as_current_span",
"style": "fstring",
"file": "invoices/worker.py",
"line": 29
}
],
"outbound": [],
"excepts": [
{
"file": "invoices/worker.py",
"line": 34,
"type": "JobFailed",
"broad": false,
"raises": false
},
{
"file": "invoices/worker.py",
"line": 52,
"type": "Exception",
"broad": true,
"raises": false
}
],
"flags": [
{
"uid": "F1",
"id": "PO-17",
"sev": "medium",
"file": "invoices/worker.py",
"line": 29,
"text": "the span name is built from data",
"detail": "Tracing backends group and sample by span name; a name per order or user makes every span unique, so latency by operation is lost. Use a constant name (\"charge_payment\") and put the value in an attribute (span.set_attribute(\"order.id\", ...))."
},
{
"uid": "F2",
"id": "PO-18",
"sev": "medium",
"file": "invoices/worker.py",
"line": 34,
"text": "the exception is handled inside the span without record_exception or set_status",
"detail": "start_as_current_span only marks a span as failed when the exception leaves the with block; caught here, the span reports success for a failed operation. Call span.record_exception(exc) and span.set_status(Status(StatusCode.ERROR))."
},
{
"uid": "F3",
"id": "PO-05",
"sev": "medium",
"file": "invoices/worker.py",
"line": 53,
"text": "the broad except logs the error without its traceback and does not re-raise",
"detail": "logger.error(...) records only the message; the stack trace that says where it failed is lost. Use logger.exception(...) or exc_info=True inside the except."
},
{
"uid": "F4",
"id": "PO-19",
"sev": "low",
"file": "invoices/worker.py",
"line": 15,
"text": "the TracerProvider has no service.name resource",
"detail": "Without a service.name every span arrives as unknown_service, so this service cannot be told apart from any other in the backend. Pass resource=Resource.create({\"service.name\": ...}) or set OTEL_SERVICE_NAME."
},
{
"uid": "F5",
"id": "PO-14",
"sev": "low",
"file": "invoices/worker.py",
"line": 23,
"text": "the Summary \"invoice_render_duration_seconds\" cannot be aggregated across replicas",
"detail": "prometheus_client's Summary exports only _count and _sum (no quantiles), and summary quantiles never add up across instances anyway. A Histogram with buckets gives percentiles over the whole fleet with histogram_quantile()."
},
{
"uid": "F6",
"id": "PO-20",
"sev": "low",
"file": "invoices/worker.py",
"line": 35,
"text": "the logs carry no trace id",
"detail": "The service traces, but no log line carries the trace or span id, so a log line cannot be opened as its trace and a slow trace cannot be read with its logs. Add the ids with LoggingInstrumentor or a structlog processor that reads trace.get_current_span()."
},
{
"uid": "F7",
"id": "PO-15",
"sev": "low",
"file": "invoices/worker.py",
"line": 38,
"text": "a duration is measured with time.time()",
"detail": "time.time() follows the wall clock, which NTP and daylight corrections move backwards and forwards; a duration from it can be negative or off by seconds. Use time.perf_counter() (or time.monotonic())."
},
{
"uid": "F8",
"id": "PO-26",
"sev": "info",
"file": "invoices/worker.py",
"line": 2,
"text": "context: no web or task framework imported; telemetry libraries structlog, prometheus_client, opentelemetry",
"detail": "Only what is pasted is checked: middleware, logging configuration or instrumentation set up in a module that was not pasted is outside the check."
}
],
"verdict_candidate": "needs_work",
"notes": []
}
The request body the page sends, trimmed for display: the full body is
7,325 characters of compact JSON, and the two long strings are cut here after their first 160
characters, with the cut marked inside the string. The step 4 samples build the untrimmed body from
worker-check.json.
{
"task": "audit",
"code": "# file: invoices/worker.py\n\"\"\"Renders monthly invoices for Quillmoor Energy customers from the billing queue.\"\"\"\nimport time\n\nimport structlog\nfrom opentelemet ... [trimmed for display: 160 of 2,018 characters shown]",
"facts": "{\"service\":{\"framework\":null,\"libraries\":[\"structlog\",\"prometheus_client\",\"opentelemetry\"],\"python_files\":1,\"handlers\":0},\"files\":[{\"path\":\"invoices/worker.py\", ... [trimmed for display: 160 of 4,574 characters shown]",
"note": "Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?"
}
The page sends it with Idempotency-Key: pyobs-desk:audit:d354df7f-1c9d:a1 (see step 5). The reply,
whole (the model’s output.output, parsed and indented here):
{
"task": "audit",
"verdict": "needs_work",
"headline": "Every render gets a span, a counter and a log line, but the two failure paths lose their traceback and the span always reports success, so a stuck or crashing worker would show almost nothing wrong.",
"service": "A queue worker (no web framework) that pops billing jobs off a BillingQueue, renders each invoice to PDF via invoices.render.render_pdf, and today emits one OpenTelemetry span per render, a Prometheus counter and duration metric for outcomes, and three structlog log lines (warning, info, error).",
"signals": [
{
"signal": "logs",
"status": "partial",
"note": "Three structured structlog calls with fields exist (no f-strings, no root logger), but the JobFailed and the broad-except log lines both drop the traceback, and the process entrypoint prints its startup line instead of logging it."
},
{
"signal": "metrics",
"status": "partial",
"note": "invoices_rendered_total and invoice_render_duration_seconds cover render outcomes, but the duration is a Summary timed with the wall clock, and a job that crashes outside render_invoice never increments any counter."
},
{
"signal": "traces",
"status": "partial",
"note": "A span wraps every render, but its name is built from the invoice id, a caught JobFailed never marks it as an error, and the TracerProvider has no service.name so spans export as unknown_service."
},
{
"signal": "correlation",
"status": "missing",
"note": "invoice_id/job_id are passed as ad-hoc keyword fields on individual log calls, but nothing binds a shared correlation or trace id that ties a job's logs, its span and any downstream call together."
}
],
"findings": [
{
"id": "PO-05",
"flags": [
"F3"
],
"file": "invoices/worker.py",
"line": 53,
"severity": "medium",
"pillar": "logs",
"title": "the broad except logs a crashed job without its traceback",
"evidence": " except Exception as exc:\n logger.error(\"job crashed\", job_id=job.id, error=str(exc))",
"why": "logger.error(...) records only the message and str(exc); the stack frame that says whether render_invoice, queue.ack or something else raised is discarded, and the except does not re-raise, so this is the only chance to see where a crash happened.",
"edits": [
{
"before": " logger.error(\"job crashed\", job_id=job.id, error=str(exc))",
"after": " logger.exception(\"job crashed\", job_id=job.id, error=str(exc))"
}
]
},
{
"id": "other",
"flags": [],
"file": "invoices/worker.py",
"line": 35,
"severity": "medium",
"pillar": "logs",
"title": "JobFailed is fully handled but logged without a traceback",
"evidence": " logger.warning(\"invoice render failed\", invoice_id=job.invoice_id, error=str(err))",
"why": "This except logs, counts and returns None instead of re-raising - it is the only record of the failure - but it only carries str(err); the traceback that shows where rendering broke inside render_pdf is gone the moment this line runs.",
"edits": [
{
"before": " logger.warning(\"invoice render failed\", invoice_id=job.invoice_id, error=str(err))",
"after": " logger.warning(\"invoice render failed\", invoice_id=job.invoice_id, error=str(err), exc_info=True)"
}
]
},
{
"id": "other",
"flags": [],
"file": "invoices/worker.py",
"line": 54,
"severity": "medium",
"pillar": "metrics",
"title": "a job that crashes outside render_invoice is never counted",
"evidence": " except Exception as exc:\n logger.error(\"job crashed\", job_id=job.id, error=str(exc))\n queue.nack(job)",
"why": "invoices_rendered_total is only touched inside render_invoice (outcome=\"ok\"/\"failed\"); if render_invoice itself raises, or queue.ack raises, this except is the only code that sees it, and nothing here increments a metric - a burst of crashing jobs would not move any number anyone is watching.",
"edits": [
{
"before": " queue.nack(job)",
"after": " INVOICES_RENDERED.labels(outcome=\"crashed\").inc()\n queue.nack(job)"
}
]
},
{
"id": "PO-18",
"flags": [
"F2"
],
"file": "invoices/worker.py",
"line": 34,
"severity": "medium",
"pillar": "traces",
"title": "the render span reports success even when JobFailed is caught",
"evidence": " except JobFailed as err:\n logger.warning(\"invoice render failed\", invoice_id=job.invoice_id, error=str(err))\n INVOICES_RENDERED.labels(outcome=\"failed\").inc()\n return None",
"why": "start_as_current_span only marks a span as an error when the exception escapes the with block; caught and swallowed here, the span for a failed render still exports as OK, so a trace backend cannot be used to find failing invoices.",
"edits": [
{
"before": " INVOICES_RENDERED.labels(outcome=\"failed\").inc()",
"after": " span.record_exception(err)\n span.set_status(trace.Status(trace.StatusCode.ERROR))\n INVOICES_RENDERED.labels(outcome=\"failed\").inc()"
}
]
},
{
"id": "PO-17",
"flags": [
"F1"
],
"file": "invoices/worker.py",
"line": 29,
"severity": "medium",
"pillar": "traces",
"title": "the span name is built from the invoice id",
"evidence": " with tracer.start_as_current_span(f\"render invoice {job.invoice_id}\") as span:",
"why": "Every invoice gets its own span name, so a tracing backend that groups or samples by name treats every render as a different operation - there is no way to see p99 render latency across invoices. The id is already available as an attribute (line 30).",
"edits": [
{
"before": " with tracer.start_as_current_span(f\"render invoice {job.invoice_id}\") as span:",
"after": " with tracer.start_as_current_span(\"render_invoice\") as span:"
}
]
},
{
"id": "PO-19",
"flags": [
"F4"
],
"file": "invoices/worker.py",
"line": 15,
"severity": "low",
"pillar": "traces",
"title": "the TracerProvider has no service.name",
"evidence": "provider = TracerProvider()",
"why": "Without a service.name resource, every span this worker exports arrives at the backend as unknown_service, indistinguishable from any other service sharing the collector.",
"edits": [
{
"before": "from opentelemetry.sdk.trace import TracerProvider",
"after": "from opentelemetry.sdk.trace import TracerProvider\nfrom opentelemetry.sdk.resources import Resource"
},
{
"before": "provider = TracerProvider()",
"after": "provider = TracerProvider(resource=Resource.create({\"service.name\": \"invoices-worker\"}))"
}
]
},
{
"id": "PO-14",
"flags": [
"F5"
],
"file": "invoices/worker.py",
"line": 23,
"severity": "low",
"pillar": "metrics",
"title": "invoice_render_duration_seconds is a Summary, not a Histogram",
"evidence": "RENDER_TIME = Summary(\"invoice_render_duration_seconds\", \"Time to render one invoice\")",
"why": "prometheus_client's Summary only exports _count and _sum for this process; its quantiles (if any were configured) never aggregate across replicas. A Histogram lets histogram_quantile() compute percentiles across the whole fleet.",
"edits": [
{
"before": "from prometheus_client import Counter, Summary",
"after": "from prometheus_client import Counter, Histogram"
},
{
"before": "RENDER_TIME = Summary(\"invoice_render_duration_seconds\", \"Time to render one invoice\")",
"after": "RENDER_TIME = Histogram(\"invoice_render_duration_seconds\", \"Time to render one invoice\")"
}
]
},
{
"id": "PO-15",
"flags": [
"F7"
],
"file": "invoices/worker.py",
"line": 38,
"severity": "low",
"pillar": "metrics",
"title": "render duration is measured with time.time()",
"evidence": " RENDER_TIME.observe(time.time() - started)",
"why": "time.time() follows the wall clock; an NTP step or DST adjustment mid-render can move it backwards, producing a negative or inflated duration recorded into invoice_render_duration_seconds.",
"edits": [
{
"before": " started = time.time()",
"after": " started = time.perf_counter()"
},
{
"before": " RENDER_TIME.observe(time.time() - started)",
"after": " RENDER_TIME.observe(time.perf_counter() - started)"
}
]
},
{
"id": "PO-20",
"flags": [
"F6"
],
"file": "invoices/worker.py",
"line": 35,
"severity": "low",
"pillar": "traces",
"title": "log lines carry no trace id",
"evidence": " logger.warning(\"invoice render failed\", invoice_id=job.invoice_id, error=str(err))",
"why": "No structlog processor here reads trace.get_current_span(), and no LoggingInstrumentor is set up, so none of the three log lines carry a trace or span id - a log line cannot be opened as its trace. There is no structlog.configure(...) call anywhere in this paste, so the processor chain that would add this is set up elsewhere in the process, outside what we can see.",
"edits": []
},
{
"id": "PO-01",
"flags": [],
"file": "invoices/worker.py",
"line": 58,
"severity": "low",
"pillar": "logs",
"title": "the startup banner uses print(), not the logger",
"evidence": " print(\"invoice worker starting\")",
"why": "print() bypasses structlog entirely, so this one line never gets the service's fields or formatting and does not ship wherever the other log lines go.",
"edits": [
{
"before": " print(\"invoice worker starting\")",
"after": " logger.info(\"worker starting\")"
}
]
}
],
"flag_answers": [
{
"flag": "F1",
"status": "confirmed",
"reason": "line 29's span name is an f-string built from job.invoice_id."
},
{
"flag": "F2",
"status": "confirmed",
"reason": "the except JobFailed block at line 34 catches and swallows the error without span.record_exception or set_status."
},
{
"flag": "F3",
"status": "confirmed",
"reason": "line 53's broad except logs via logger.error with no traceback and no re-raise."
},
{
"flag": "F4",
"status": "confirmed",
"reason": "line 15 builds TracerProvider() with no resource argument."
},
{
"flag": "F5",
"status": "confirmed",
"reason": "line 23 declares RENDER_TIME as a Summary, not a Histogram."
},
{
"flag": "F6",
"status": "confirmed",
"reason": "none of the three structlog calls carry a trace or span id; no LoggingInstrumentor or trace-reading processor is present."
},
{
"flag": "F7",
"status": "confirmed",
"reason": "lines 31 and 38 both use time.time() to measure the render duration."
}
],
"plan": [
"Add the traceback to the broad except's crash log (line 53) so a crash can be diagnosed.",
"Add exc_info=True to the JobFailed log (line 35) - it is the only record of that failure.",
"Count crashed jobs (line 54) so a crash burst moves a metric, not just a log line.",
"Mark the span as an error when JobFailed is caught (line 34-36).",
"Give the span a constant name and keep the invoice id in the attribute already set (line 29).",
"Add a service.name resource to the TracerProvider (line 15).",
"Switch invoice_render_duration_seconds from Summary to Histogram (line 23).",
"Time the render with time.perf_counter() instead of time.time() (lines 31, 38).",
"Replace the startup print() with a logger.info() call (line 58).",
"Decide, outside telemetry, whether a structlog processor or LoggingInstrumentor should stamp trace ids on logs (PO-20)."
],
"questions": [
"render_invoice returns None when JobFailed is caught (line 37), but run_forever calls queue.ack(job) right after render_invoice (lines 50-51) with no check of that return value, so a failed render is acked off the queue exactly like a successful one. That looks like the real reason invoices go missing - it is a control-flow/return-value question, not a telemetry gap, so no edit was made for it: should a JobFailed render be nacked or retried instead of acked?",
"Is a structlog processor or LoggingInstrumentor already configured elsewhere in the process to stamp trace/span ids onto log lines? None is visible in this paste, so PO-20 was left without an edit.",
"BillingQueue.pop(timeout=30) implies a backlog; is there already a queue-depth or backlog gauge elsewhere? If not, that is a saturation signal worth adding alongside these fixes."
],
"summary": "The two places a render can fail don't leave a usable trail: the JobFailed except logs only str(err) with no traceback, and a job that crashes outside render_invoice is logged (also without a traceback) but never counted, so a crash burst would not move invoices_rendered_total at all. The span makes it worse - it is named per invoice, so nothing aggregates by operation, and it is never marked as an error when JobFailed is caught, so a failed render still shows a healthy span. Separately, and not a telemetry issue: run_forever acks the job right after calling render_invoice regardless of whether it returned a PDF or None, so a JobFailed render is permanently removed from the queue with nothing beyond a WARNING log and an outcome=\"failed\" counter tick - if nobody is watching that counter, the invoice is gone for good and none of today's telemetry would tell you; see the first question."
}
Held to the page’s own reconciliation (recon.js reconcile, on this
exact body and reply), this reply has 0 disagreements in 8 statements
checked: the four signals are rated once each, every one of the 10 findings names a pasted file, a
line inside it and evidence copied from it, the verdict fits the worst finding, all 7 flags are
answered once, and all 12 edits apply in order with each before found exactly once. With
every edit applied the free check raises 0 high, 0 medium and 1 low flag: PO-20 (the logs carry no
trace id), whose finding carries no edit because the structlog configuration is outside the paste
— as the contract allows for a low finding. The recheck.js script under
The page checks the reply against the code runs that reconciliation for you.
What the step 5 samples print for this reply (the Python form; charged is whatever the run cost):
truncated: False charged: ...
needs_work - Every render gets a span, a counter and a log line, but the two failure paths lose their traceback and the span always reports success, so a stuck or crashing worker would show almost nothing wrong.
signal logs partial
signal metrics partial
signal traces partial
signal correlation missing
PO-05 invoices/worker.py 53 medium 1 edit(s)
other invoices/worker.py 35 medium 1 edit(s)
other invoices/worker.py 54 medium 1 edit(s)
PO-18 invoices/worker.py 34 medium 1 edit(s)
PO-17 invoices/worker.py 29 medium 1 edit(s)
PO-19 invoices/worker.py 15 low 2 edit(s)
PO-14 invoices/worker.py 23 low 2 edit(s)
PO-15 invoices/worker.py 38 low 2 edit(s)
PO-20 invoices/worker.py 35 low 0 edit(s)
PO-01 invoices/worker.py 58 low 1 edit(s)
answer F1 confirmed
answer F2 confirmed
answer F3 confirmed
answer F4 confirmed
answer F5 confirmed
answer F6 confirmed
answer F7 confirmed
Had you pressed “Apply every edit and instrument the rest” after this reply, the page
would put the 12 edits into the paste (63 lines now), recompute facts from the edited
paste, and send the instrument lane (with the same note) with this audit_handoff —
what handoff.js prints as audit_handoff — shown decoded and
trimmed to 2 of its 10 findings; in the body it is the compact
JSON.stringify of it (4,142 characters). That body is 9,809 characters and its
key would be pyobs-desk:instrument:697f83b3-2651:a1:
{
"verdict": "needs_work",
"headline": "Every render gets a span, a counter and a log line, but the two failure paths lose their traceback and the span always reports success, so a stuck or crashing worker would show almost nothing wrong.",
"findings": [
{
"id": "PO-05",
"severity": "medium",
"file": "invoices/worker.py",
"line": 53,
"title": "the broad except logs a crashed job without its traceback",
"why": "logger.error(...) records only the message and str(exc); the stack frame that says whether render_invoice, queue.ack or something else raised is discarded, and the except does not re-raise, so this is the only chance to see where a crash happened.",
"edits": 1
},
{
"id": "other",
"severity": "medium",
"file": "invoices/worker.py",
"line": 35,
"title": "JobFailed is fully handled but logged without a traceback",
"why": "This except logs, counts and returns None instead of re-raising - it is the only record of the failure - but it only carries str(err); the traceback that shows where rendering broke inside render_pdf is gone the moment this line runs.",
"edits": 1
},
... [trimmed for display: 8 more findings, same shape]
],
"applied": 12,
"failed": 0
}
Instrument: a shipping quote API with one log line and no metrics or traces
One file, quotes/api.py: a FastAPI service with two handlers (POST /quotes
and GET /lanes/{origin}), one outbound httpx.get to a carrier, stdlib
logging and nothing else. The free check raises 2 flags (2 medium); its reading of the
verdict is needs_work. This example has no audit before it, so it carries no
audit_handoff. The engineer’s note: “Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.”
The code string, whole: 1,430 characters, 45 lines ending in a
newline. Save it as quotes-paste.txt and run
node build-check.js quotes-paste.txt > quotes-check.json:
# file: quotes/api.py
"""Shipping quote API for Harbour & Pine Freight. Returns a price for a parcel between two postcodes."""
import logging
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from quotes.rates import RateTable, UnknownLane
logger = logging.getLogger(__name__)
app = FastAPI(title="Harbour & Pine quotes")
rates = RateTable.load("rates.csv")
CARRIER_URL = "https://carrier-rates.harbourpine.example/v1/surcharges"
class QuoteRequest(BaseModel):
origin: str
destination: str
weight_kg: float
express: bool = False
def fuel_surcharge(origin: str, destination: str) -> float:
resp = httpx.get(CARRIER_URL, params={"from": origin, "to": destination}, timeout=3)
resp.raise_for_status()
return float(resp.json()["fuel_pct"])
@app.post("/quotes")
def create_quote(req: QuoteRequest):
try:
base = rates.price(req.origin, req.destination, req.weight_kg)
except UnknownLane:
raise HTTPException(status_code=422, detail="no rate for this lane")
surcharge = fuel_surcharge(req.origin, req.destination)
total = round(base * (1 + surcharge / 100) * (1.6 if req.express else 1.0), 2)
logger.info("quote created")
return {"price": total, "currency": "EUR", "express": req.express}
@app.get("/lanes/{origin}")
def list_lanes(origin: str):
return {"origin": origin, "destinations": rates.destinations(origin)}
The facts object, decoded and whole (2,186 characters as the page’s string):
{
"service": {
"framework": "FastAPI",
"libraries": [
"logging"
],
"python_files": 1,
"handlers": 2
},
"files": [
{
"path": "quotes/api.py",
"kind": "python",
"line": 2,
"end_line": 46
}
],
"handlers": [
{
"name": "create_quote",
"file": "quotes/api.py",
"line": 32,
"kind": "http",
"method": "POST",
"route": "/quotes"
},
{
"name": "list_lanes",
"file": "quotes/api.py",
"line": 44,
"kind": "http",
"method": "GET",
"route": "/lanes/{origin}"
}
],
"functions": [
{
"name": "fuel_surcharge",
"file": "quotes/api.py",
"line": 25
},
{
"name": "create_quote",
"file": "quotes/api.py",
"line": 32
},
{
"name": "list_lanes",
"file": "quotes/api.py",
"line": 44
}
],
"logs": [
{
"file": "quotes/api.py",
"line": 39,
"level": "info",
"lib": "stdlib",
"style": "literal",
"message": "quote created",
"fields": [],
"exc": false
}
],
"metrics": [],
"spans": [],
"outbound": [
{
"file": "quotes/api.py",
"line": 26,
"call": "httpx.get"
}
],
"excepts": [
{
"file": "quotes/api.py",
"line": 35,
"type": "UnknownLane",
"broad": false,
"raises": true
}
],
"flags": [
{
"uid": "F1",
"id": "PO-22",
"sev": "medium",
"file": "quotes/api.py",
"line": 32,
"text": "no correlation or request id is attached to the logs",
"detail": "With 2 request handlers writing logs, nothing ties the lines of one request together: under concurrency the lines of different requests interleave and cannot be separated. Set an id per request in middleware (a contextvar bound into every log line) and return it in a response header."
},
{
"uid": "F2",
"id": "PO-24",
"sev": "medium",
"file": "quotes/api.py",
"line": 32,
"text": "2 handlers have no latency or traffic signal: create_quote, list_lanes",
"detail": "Latency, traffic and errors - three of the four golden signals - are measured per handler, and nothing here records them: no metric, no span, no instrumentation middleware. Time each handler with a Histogram labelled by route template, method and status (or install the framework's instrumentation).",
"lines": [
32,
44
]
},
{
"uid": "F3",
"id": "PO-26",
"sev": "info",
"file": "quotes/api.py",
"line": 2,
"text": "context: FastAPI; telemetry libraries logging",
"detail": "Also read: framework: FastAPI. Only what is pasted is checked: middleware, logging configuration or instrumentation set up in a module that was not pasted is outside the check."
}
],
"verdict_candidate": "needs_work",
"notes": []
}
The request body, trimmed for display the same way (the full body is 4,185
characters), sent with Idempotency-Key: pyobs-desk:instrument:713237d2-1059:a1:
{
"task": "instrument",
"code": "# file: quotes/api.py\n\"\"\"Shipping quote API for Harbour & Pine Freight. Returns a price for a parcel between two postcodes.\"\"\"\nimport logging\n\nimport httpx\nfrom ... [trimmed for display: 160 of 1,430 characters shown]",
"facts": "{\"service\":{\"framework\":\"FastAPI\",\"libraries\":[\"logging\"],\"python_files\":1,\"handlers\":2},\"files\":[{\"path\":\"quotes/api.py\",\"kind\":\"python\",\"line\":2,\"end_line\":46 ... [trimmed for display: 160 of 2,186 characters shown]",
"note": "Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API."
}
The reply, parsed and indented, with the one file’s content
cut after 200 characters (marked inside the string); every other value is whole:
{
"task": "instrument",
"verdict": "instrumented",
"headline": "quotes/api.py now carries a per-request correlation id, a method/route/status latency-and-traffic histogram, and OpenTelemetry spans (with trace-context and correlation propagation on the outbound carrier call).",
"files": [
{
"path": "quotes/api.py",
"content": "\"\"\"Shipping quote API for Harbour & Pine Freight. Returns a price for a parcel between two postcodes.\"\"\"\nimport logging\nimport time\nimport uuid\nfrom contextvars import ContextVar\n\nimport httpx\nfrom fa ... [trimmed for display: 200 of 4,791 characters shown; the whole file is below]"
}
],
"signals": [
{
"kind": "metric",
"name": "http_request_duration_seconds",
"type": "histogram",
"labels": [
"method",
"route",
"status"
],
"where": "correlation_middleware",
"why": "Times every request to create_quote and list_lanes and records its outcome, labelled by method, the matched route template (or the constant \"unmatched\") and status, so latency, traffic and errors are visible per route with a bounded label set - this closes F2 (PO-24)."
},
{
"kind": "span",
"name": "http_request",
"type": "span",
"labels": [],
"where": "correlation_middleware",
"why": "Root span for the whole request/response cycle so a failure anywhere inside create_quote, fuel_surcharge or list_lanes (including one that escapes uncaught) is recorded with the matched route and status code."
},
{
"kind": "span",
"name": "create_quote",
"type": "span",
"labels": [],
"where": "create_quote",
"why": "Traces the rate lookup, fuel surcharge call and pricing for POST /quotes; the UnknownLane except is inside this span and explicitly records the exception since it is handled (turned into a 422) rather than left to propagate."
},
{
"kind": "span",
"name": "fuel_surcharge_lookup",
"type": "span",
"labels": [],
"where": "fuel_surcharge",
"why": "Traces the one outbound dependency (the httpx.get to CARRIER_URL); any httpx exception propagates out of this span uncaught, so it is recorded and marked as an error automatically."
},
{
"kind": "span",
"name": "list_lanes",
"type": "span",
"labels": [],
"where": "list_lanes",
"why": "Traces GET /lanes/{origin} so its RateTable.destinations call appears in the same trace as the rest of the service."
},
{
"kind": "log",
"name": "request_failed",
"type": "log",
"labels": [],
"where": "correlation_middleware",
"why": "Logs the traceback (exc_info=True) for any exception that escapes a handler - for example an httpx failure inside fuel_surcharge - before it becomes an unhandled 500; nothing logged this path before."
},
{
"kind": "log",
"name": "quote_lane_unknown",
"type": "log",
"labels": [],
"where": "create_quote",
"why": "Logs the UnknownLane case with its traceback at WARNING, not ERROR, since a 422 is an expected client error (PO-08); previously this branch produced no log line at all."
}
],
"setup": [
"Add prometheus_client and opentelemetry-api to the service's dependencies (pip install prometheus_client opentelemetry-api).",
"At process start, create a TracerProvider with Resource.create({\"service.name\": \"quotes-api\"}) (or set OTEL_SERVICE_NAME) and register a span processor with an OTLP exporter pointed at the collector in front of Tempo - this needs opentelemetry-sdk and an OTLP exporter package; this module only calls trace.get_tracer(__name__) and never configures the provider itself.",
"Configure the logging format/handlers (in the process or deployment, not in this module) so %(request_id)s and %(trace_id)s - already attached to every record by RequestContextFilter - actually appear in the emitted line, for example '%(asctime)s %(levelname)s %(name)s request_id=%(request_id)s trace_id=%(trace_id)s %(message)s'.",
"Scrape the /metrics endpoint this module now mounts (prometheus_client.make_asgi_app()) with Prometheus, or route it through whatever already reaches Tempo for traces.",
"If a gateway or load balancer sits in front of this service, have it forward the same X-Request-ID (or X-Correlation-ID) header name it already sets, so correlation ids are carried end-to-end instead of being regenerated at this hop."
],
"open": [],
"questions": [
"origin/destination are postcodes; this instrumentation keeps them out of every log field and span attribute on the assumption they count as address-level personal data under the house rule - confirm whether they may be attached (for example as span attributes on create_quote) to make a specific quote easier to debug, or whether they should stay redacted as they are now.",
"fuel_surcharge propagates trace context and the correlation id with a manual inject(headers) call rather than HTTPXClientInstrumentor, per the note that only prometheus_client and opentelemetry-api should be added - confirm that's still preferred over adding opentelemetry-instrumentation-httpx as a dependency."
],
"summary": "Every handler now has a latency/traffic signal, a correlation id set in middleware and returned on the response, and OpenTelemetry spans around each handler's work and around the one outbound call, with the trace context and the correlation id both sent to the carrier; this closes F1 (PO-22) and F2 (PO-24) directly. What the note asked for - Prometheus and an OpenTelemetry/Tempo collector - is wired into the code (Histogram, /metrics mount, tracer, propagation); the TracerProvider/exporter, log formatting and OTEL_SERVICE_NAME are process-start configuration outside this paste and are listed in setup rather than assumed."
}
The content of quotes/api.py in that reply, whole (4,791 characters):
"""Shipping quote API for Harbour & Pine Freight. Returns a price for a parcel between two postcodes."""
import logging
import time
import uuid
from contextvars import ContextVar
import httpx
from fastapi import FastAPI, HTTPException, Request
from opentelemetry import trace
from opentelemetry.propagate import inject
from opentelemetry.trace import Status, StatusCode
from prometheus_client import Histogram, make_asgi_app
from pydantic import BaseModel
from quotes.rates import RateTable, UnknownLane
logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)
app = FastAPI(title="Harbour & Pine quotes")
rates = RateTable.load("rates.csv")
CARRIER_URL = "https://carrier-rates.harbourpine.example/v1/surcharges"
request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
class RequestContextFilter(logging.Filter):
"""Attach the current request id and trace id to every log record."""
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id_var.get()
span_context = trace.get_current_span().get_span_context()
record.trace_id = format(span_context.trace_id, "032x") if span_context.is_valid else "-"
return True
logger.addFilter(RequestContextFilter())
HTTP_REQUEST_DURATION_SECONDS = Histogram(
"http_request_duration_seconds",
"Duration of HTTP requests in seconds, labelled by method, route template and status",
["method", "route", "status"],
)
app.mount("/metrics", make_asgi_app())
class QuoteRequest(BaseModel):
origin: str
destination: str
weight_kg: float
express: bool = False
@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
request_id = (
request.headers.get("X-Request-ID")
or request.headers.get("X-Correlation-ID")
or str(uuid.uuid4())
)
token = request_id_var.set(request_id)
start = time.perf_counter()
try:
with tracer.start_as_current_span("http_request") as span:
span.set_attribute("http.method", request.method)
try:
response = await call_next(request)
except Exception as exc:
route = request.scope.get("route")
route_template = route.path if route is not None else "unmatched"
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR))
logger.error(
"request_failed",
exc_info=True,
extra={"method": request.method, "route": route_template},
)
HTTP_REQUEST_DURATION_SECONDS.labels(
method=request.method, route=route_template, status="500"
).observe(time.perf_counter() - start)
raise
route = request.scope.get("route")
route_template = route.path if route is not None else "unmatched"
status = str(response.status_code)
HTTP_REQUEST_DURATION_SECONDS.labels(
method=request.method, route=route_template, status=status
).observe(time.perf_counter() - start)
span.set_attribute("http.route", route_template)
span.set_attribute("http.status_code", response.status_code)
response.headers["X-Request-ID"] = request_id
return response
finally:
request_id_var.reset(token)
def fuel_surcharge(origin: str, destination: str) -> float:
headers = {"X-Request-ID": request_id_var.get()}
inject(headers)
with tracer.start_as_current_span("fuel_surcharge_lookup"):
resp = httpx.get(
CARRIER_URL, params={"from": origin, "to": destination}, timeout=3, headers=headers
)
resp.raise_for_status()
return float(resp.json()["fuel_pct"])
@app.post("/quotes")
def create_quote(req: QuoteRequest):
with tracer.start_as_current_span("create_quote") as span:
try:
base = rates.price(req.origin, req.destination, req.weight_kg)
except UnknownLane as exc:
span.record_exception(exc)
span.set_status(Status(StatusCode.ERROR))
logger.warning("quote_lane_unknown", exc_info=True)
raise HTTPException(status_code=422, detail="no rate for this lane")
surcharge = fuel_surcharge(req.origin, req.destination)
total = round(base * (1 + surcharge / 100) * (1.6 if req.express else 1.0), 2)
logger.info("quote created")
return {"price": total, "currency": "EUR", "express": req.express}
@app.get("/lanes/{origin}")
def list_lanes(origin: str):
with tracer.start_as_current_span("list_lanes"):
return {"origin": origin, "destinations": rates.destinations(origin)}
Held to the page’s reconciliation, this reply has 0 disagreements in
7 statements checked: the one file comes back whole with no tabs; with it put into the paste
the free check raises no high or medium flag (and no low one either); all 3 functions and both
handlers keep their names, methods and routes; and signals matches the code — 1
metric (http_request_duration_seconds, a Histogram labelled method, route and status)
and 4 span names. The verdict instrumented fits the empty open list.
What the step 6 samples print for this reply (the Python form):
truncated: False charged: ...
instrumented - quotes/api.py now carries a per-request correlation id, a method/route/status latency-and-traffic histogram, and OpenTelemetry spans (with trace-context and correlation propagation on the outbound carrier call).
file quotes/api.py 4791 characters
signal metric http_request_duration_seconds histogram method,route,status in correlation_middleware
signal span http_request span - in correlation_middleware
signal span create_quote span - in create_quote
signal span fuel_surcharge_lookup span - in fuel_surcharge
signal span list_lanes span - in list_lanes
signal log request_failed log - in correlation_middleware
signal log quote_lane_unknown log - in create_quote
setup Add prometheus_client and opentelemetry-api to the service's dependencies (pip install prometheus_client opentelemetry-api).
setup At process start, create a TracerProvider with Resource.create({"service.name": "quotes-api"}) (or set OTEL_SERVICE_NAME) and register a span processor with an OTLP exporter pointed at the collector in front of Tempo - this needs opentelemetry-sdk and an OTLP exporter package; this module only calls trace.get_tracer(__name__) and never configures the provider itself.
setup Configure the logging format/handlers (in the process or deployment, not in this module) so %(request_id)s and %(trace_id)s - already attached to every record by RequestContextFilter - actually appear in the emitted line, for example '%(asctime)s %(levelname)s %(name)s request_id=%(request_id)s trace_id=%(trace_id)s %(message)s'.
setup Scrape the /metrics endpoint this module now mounts (prometheus_client.make_asgi_app()) with Prometheus, or route it through whatever already reaches Tempo for traces.
setup If a gateway or load balancer sits in front of this service, have it forward the same X-Request-ID (or X-Correlation-ID) header name it already sets, so correlation ids are carried end-to-end instead of being regenerated at this hop.
question origin/destination are postcodes; this instrumentation keeps them out of every log field and span attribute on the assumption they count as address-level personal data under the house rule - confirm whether they may be attached (for example as span attributes on create_quote) to make a specific quote easier to debug, or whether they should stay redacted as they are now.
question fuel_surcharge propagates trace context and the correlation id with a manual inject(headers) call rather than HTTPXClientInstrumentor, per the note that only prometheus_client and opentelemetry-api should be added - confirm that's still preferred over adding opentelemetry-instrumentation-httpx as a dependency.
The page checks the reply against the code; the API does not
What comes back from the API is the model’s raw text in output.output. The page
holds it to account in the browser before showing it (recon.js
reconcile, with the facts, the code and the handoff it sent): it applies every edit or
returned file to the paste and re-runs its own checks on the result. None of that happens in the API.
A direct API caller gets the reply unchecked and must do the applying and re-checking itself.
- Both lanes: the reply is the lane you asked for (one written for the other task is a disagreement), it carries a valid verdict for that lane, and no credential from the paste is repeated anywhere in it (every credential value the free check found in the paste).
- Audit: the signals table. Exactly four rows, covering logs, metrics, traces and correlation once each.
- Audit: every finding is anchored. Its
idis in the catalogue (PO-01..PO-26, orother; PO-26 is context and never a finding); itsfileis a pasted file; itslineis inside that file; itsevidenceis found line by line in that file ofcode(indentation trimmed; a line containing***is skipped) and on or within three lines ofline; every citeduidis a non-info flag the page raised. A finding that carries a high flag at a lower severity, or a medium or high finding with no edit, is pointed out. - Audit: every flag is answered once, consistently. Every non-info flag answered exactly once and no other; a
confirmedanswer needs a finding carrying that uid; adismissedone must not be carried by a finding and should give a reason. The verdict must fit the worst severity (blockedwith a high finding,needs_workwith a medium, otherwiseready). - Audit: every edit is put in place and re-checked. The findings’ edits are applied in the listed order, each on top of the ones before (
Obskit.applyEdits): abeforethat is not found, is found more than once, or changes nothing is a disagreement. After each finding the free check runs again: the code must still read and no new high flag may appear. With every edit applied, every confirmed flag a finding with edits carries must be gone, every function of the paste must still exist, and a new medium flag is pointed out. - Instrument: the files. At least one file, at most one new one; none empty, none with a tab or a
# file:line, every path a.pyfile. - Instrument: put into the paste and re-checked. The files replace the pasted files of the same path (or are added) (
Obskit.replaceFiles) and the free check runs again: the code must read, no high or medium flag may remain, every function must still exist, and every handler must keep its method and route. - Instrument: the signals list matches the code. Every metric the code creates is listed with the same name (a
_totalsuffix is ignored), type and labels, every listed metric is in the code, and every constant span name is listed and opened; handlers with no metric at all are a disagreement. The verdict must fit theopenlist, and an emptysetupis pointed out.
recon.js also runs in Node, so you can hold a reply to exactly the page’s standard.
Save /obskit.js and /recon.js next to this script,
pass the file you built the body from (build-check.js or handoff.js output),
the reply text and the lane; it prints every statement, the count, and the paste with every edit (or
file) put back:
// recheck.js -- run: node recheck.js worker-check.json worker-reply.txt audit > worker-edited.txt (obskit.js, recon.js next to it)
const fs = require("fs");
const Recon = require("./recon.js"); // loads ./obskit.js itself
const sent = JSON.parse(fs.readFileSync(process.argv[2], "utf8")); // code, facts (and audit_handoff) you sent
const lane = process.argv[4] || "audit";
const res = Recon.normalize(Recon.parseResult(fs.readFileSync(process.argv[3], "utf8")), lane); // output.output
const r = Recon.reconcile(res, { facts: sent.facts, code: sent.code, handoff: sent.audit_handoff || null });
r.notes.forEach((n) => console.error(n.kind + ": " + n.text));
console.error(r.disagreements + " disagreement(s) in " + r.statements + " statements checked");
if (r.after) (r.after.flags || []).forEach((f) => console.error("still raised: " + f.id + " " + f.sev + " " + f.file + ":" + f.line + " " + f.text));
if (r.after && r.after.text) console.log(r.after.text.replace(/\n$/, "")); // the paste with every edit (or file) put back
For the worker audit reply it reports:
ok: The signals table covers logs, metrics, traces and correlation once each.
ok: Every one of the 10 findings names a pasted file, a line inside it and evidence copied from it.
ok: The verdict (needs_work) fits the worst finding.
ok: Every one of the page's 7 flags is answered once.
ok: All 12 edits applied in order, each before text found exactly once.
ok: Every confirmed flag a finding carries is gone once all edits are applied.
ok: Every function of the paste is still there after the edits.
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 8 statements checked
still raised: PO-20 low invoices/worker.py:36 the logs carry no trace id
and for the quotes instrument reply:
ok: 1 file returned whole, no tabs.
ok: With the instrumented files in place the free check raises no high or medium flag (0 low).
ok: Every one of the 3 functions is still there.
ok: Every handler keeps its method and route.
ok: signals matches the code: 1 metric (names, types and labels) and 4 span name(s).
ok: The verdict (instrumented) fits the open list.
ok: The reply repeats no credential from the paste.
0 disagreement(s) in 7 statements checked
After that, run the edited service’s own tests, start it once so every metric is registered
(a duplicate registration raises at import), scrape /metrics and look for the spans in
your tracing backend before you ship anything.
Step 1 — a tiny client helper
One function per language that does what sdk.js does for every call: JSON in,
bearer token if you have one, data out on a 2xx, an error carrying
status, error.code and error.message otherwise. The later
steps assume this helper is in the same file (for Go, Java and C#, the later snippets go inside
main / the top-level statements; Java’s makeInput from step 4 goes
in the class).
BASE="https://api.skillsafe.ai/v1/app-api"
TOKEN="${SKILLSAFE_TOKEN:-YOUR_TOKEN}" # step 2
# api METHOD PATH [extra curl args...] -- prints the whole envelope; pipe to jq
api() {
method="$1"; path="$2"; shift 2
curl -sS -X "$method" "$BASE$path" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $TOKEN" "$@"
}
# success: jq '.data' failure: jq '.error | {code, message}'
import hashlib, json, os, time, urllib.parse
import requests # pip install requests
BASE = "https://api.skillsafe.ai/v1/app-api"
TOKEN = os.environ.get("SKILLSAFE_TOKEN", "YOUR_TOKEN") # step 2
class ApiError(Exception):
def __init__(self, status, code, message, details=None):
super().__init__(f"{status} {code}: {message}")
self.status, self.code, self.details = status, code, details
def api(method, path, body=None, headers=None):
h = {"Content-Type": "application/json"}
if TOKEN:
h["Authorization"] = "Bearer " + TOKEN
h.update(headers or {})
r = requests.request(method, BASE + path, headers=h,
data=None if body is None else json.dumps(body))
try:
env = r.json()
except ValueError:
env = {}
if not r.ok:
err = env.get("error") or {}
raise ApiError(r.status_code, err.get("code"), err.get("message") or r.reason, err.get("details"))
return env.get("data")
// pyobs-desk.mjs -- Node 18+ (global fetch), run with: node pyobs-desk.mjs
import { readFileSync } from "node:fs";
import { createHash } from "node:crypto";
const BASE = "https://api.skillsafe.ai/v1/app-api";
let TOKEN = "YOUR_TOKEN"; // step 2
async function api(method, path, body, headers = {}) {
const h = { "Content-Type": "application/json", ...headers };
if (TOKEN) h.Authorization = "Bearer " + TOKEN;
const res = await fetch(BASE + path, {
method,
headers: h,
body: body === undefined ? undefined : JSON.stringify(body),
});
const json = await res.json().catch(() => ({}));
if (!res.ok) {
const err = new Error((json.error && json.error.message) || res.statusText);
err.status = res.status;
err.code = json.error && json.error.code;
err.details = json.error && json.error.details;
throw err;
}
return json.data;
}
package main
import (
"bufio"
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/url"
"os"
"strings"
"time"
)
const base = "https://api.skillsafe.ai/v1/app-api"
var token = os.Getenv("SKILLSAFE_TOKEN") // step 2
type apiError struct {
Status int
Code string
Message string
}
func (e *apiError) Error() string { return fmt.Sprintf("%d %s: %s", e.Status, e.Code, e.Message) }
type envelope struct {
OK bool `json:"ok"`
Data json.RawMessage `json:"data"`
Error *struct {
Code string `json:"code"`
Message string `json:"message"`
} `json:"error"`
}
func newRequest(method, path string, body any, headers map[string]string) (*http.Request, error) {
var buf []byte
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return nil, err
}
buf = b
}
req, err := http.NewRequest(method, base+path, bytes.NewReader(buf))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if token != "" {
req.Header.Set("Authorization", "Bearer "+token)
}
for k, v := range headers {
req.Header.Set(k, v)
}
return req, nil
}
func errorFrom(res *http.Response, env envelope) error {
e := &apiError{Status: res.StatusCode, Message: http.StatusText(res.StatusCode)}
if env.Error != nil {
e.Code = env.Error.Code
if env.Error.Message != "" {
e.Message = env.Error.Message
}
}
return e
}
// api sends body as JSON and decodes the envelope's data into out (out may be nil).
func api(method, path string, body any, headers map[string]string, out any) error {
req, err := newRequest(method, path, body, headers)
if err != nil {
return err
}
res, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
var env envelope
_ = json.NewDecoder(res.Body).Decode(&env)
if res.StatusCode < 200 || res.StatusCode > 299 {
return errorFrom(res, env)
}
if out != nil && len(env.Data) > 0 {
return json.Unmarshal(env.Data, out)
}
return nil
}
func main() {
// steps 2-6 go here; remove any import a step you skip no longer uses
_, _, _, _, _ = bufio.NewScanner, sha256.Sum256, hex.EncodeToString, url.PathEscape, strings.Index
_ = time.Second
}
// PyObsDesk.java -- Java 17+, needs com.fasterxml.jackson.core:jackson-databind on the classpath
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.URI;
import java.net.URLEncoder;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class PyObsDesk {
static final String BASE = "https://api.skillsafe.ai/v1/app-api";
static String token = System.getenv().getOrDefault("SKILLSAFE_TOKEN", "YOUR_TOKEN"); // step 2
static final HttpClient HTTP = HttpClient.newHttpClient();
static final ObjectMapper JSON = new ObjectMapper();
static class ApiException extends Exception {
final int status;
final String code;
ApiException(int status, String code, String message) {
super(status + " " + code + ": " + message);
this.status = status;
this.code = code;
}
}
static HttpRequest.Builder request(String method, String path, Object body, Map<String, String> headers) throws Exception {
HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(BASE + path))
.header("Content-Type", "application/json")
.method(method, body == null ? HttpRequest.BodyPublishers.noBody()
: HttpRequest.BodyPublishers.ofString(JSON.writeValueAsString(body)));
if (token != null && !token.isEmpty()) b.header("Authorization", "Bearer " + token);
if (headers != null) headers.forEach(b::header);
return b;
}
static ApiException errorFrom(int status, JsonNode env) {
JsonNode err = env.path("error");
return new ApiException(status, err.path("code").asText(null), err.path("message").asText("HTTP " + status));
}
static JsonNode api(String method, String path, Object body, Map<String, String> headers) throws Exception {
HttpResponse<String> res = HTTP.send(request(method, path, body, headers).build(), HttpResponse.BodyHandlers.ofString());
JsonNode env;
try { env = JSON.readTree(res.body()); } catch (Exception e) { env = JSON.createObjectNode(); }
if (env == null) env = JSON.createObjectNode();
if (res.statusCode() / 100 != 2) throw errorFrom(res.statusCode(), env);
return env.path("data");
}
public static void main(String[] args) throws Exception {
// steps 2-6 go here (makeInput from step 4 goes next to api())
}
}
require "digest"
require "json"
require "net/http"
require "uri"
BASE = "https://api.skillsafe.ai/v1/app-api"
$token = ENV.fetch("SKILLSAFE_TOKEN", "YOUR_TOKEN") # step 2
class ApiError < StandardError
attr_reader :status, :code
def initialize(status, code, message)
super("#{status} #{code}: #{message}")
@status = status
@code = code
end
end
def build_request(method, path, body = nil, headers = {})
uri = URI(BASE + path)
req = Net::HTTPGenericRequest.new(method, !body.nil?, true, uri.request_uri)
req["Content-Type"] = "application/json"
req["Authorization"] = "Bearer #{$token}" if $token && !$token.empty?
headers.each { |k, v| req[k] = v }
req.body = JSON.generate(body) unless body.nil?
[uri, req]
end
def api(method, path, body = nil, headers = {})
uri, req = build_request(method, path, body, headers)
res = Net::HTTP.start(uri.host, uri.port, use_ssl: true) { |h| h.request(req) }
env = (JSON.parse(res.body) rescue {})
env = {} unless env.is_a?(Hash)
unless res.is_a?(Net::HTTPSuccess)
err = env["error"] || {}
raise ApiError.new(res.code.to_i, err["code"], err["message"] || res.message)
end
env["data"]
end
<?php
// needs the curl extension
const BASE = "https://api.skillsafe.ai/v1/app-api";
$TOKEN = getenv("SKILLSAFE_TOKEN") ?: "YOUR_TOKEN"; // step 2
class ApiError extends Exception {
public $status;
public $apiCode;
public function __construct($status, $code, $message) {
parent::__construct("$status $code: $message");
$this->status = $status;
$this->apiCode = $code;
}
}
function headers_for(array $extra = []): array {
global $TOKEN;
$h = ["Content-Type: application/json"];
if ($TOKEN) $h[] = "Authorization: Bearer $TOKEN";
foreach ($extra as $k => $v) $h[] = "$k: $v";
return $h;
}
function api(string $method, string $path, $body = null, array $headers = []) {
$ch = curl_init(BASE . $path);
curl_setopt_array($ch, [
CURLOPT_CUSTOMREQUEST => $method,
CURLOPT_HTTPHEADER => headers_for($headers),
CURLOPT_RETURNTRANSFER => true,
]);
if ($body !== null) curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($body, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE));
$raw = curl_exec($ch);
if ($raw === false) throw new RuntimeException(curl_error($ch));
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);
$env = json_decode($raw);
if (!is_object($env)) $env = new stdClass();
if ($status < 200 || $status > 299) {
$e = $env->error ?? new stdClass();
throw new ApiError($status, $e->code ?? null, $e->message ?? "HTTP $status");
}
return $env->data ?? null;
}
// Program.cs -- .NET 6+ console app (implicit usings on) with top-level statements; steps 2-6 (with the MakeInput local function) go ABOVE this class
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
static class Api
{
public const string Base = "https://api.skillsafe.ai/v1/app-api";
public static string Token = Environment.GetEnvironmentVariable("SKILLSAFE_TOKEN") ?? "YOUR_TOKEN"; // step 2
public static readonly HttpClient Http = new HttpClient();
public class ApiException : Exception
{
public int Status;
public string? Code;
public ApiException(int status, string? code, string message) : base($"{status} {code}: {message}")
{
Status = status;
Code = code;
}
}
public static HttpRequestMessage Request(string method, string path, object? body = null, IDictionary<string, string>? headers = null)
{
var req = new HttpRequestMessage(new HttpMethod(method), Base + path);
if (body != null) req.Content = new StringContent(JsonSerializer.Serialize(body), Encoding.UTF8, "application/json");
if (!string.IsNullOrEmpty(Token)) req.Headers.TryAddWithoutValidation("Authorization", "Bearer " + Token);
if (headers != null) foreach (var kv in headers) req.Headers.TryAddWithoutValidation(kv.Key, kv.Value);
return req;
}
public static ApiException ErrorFrom(HttpResponseMessage res, JsonNode? env)
{
var err = env?["error"];
return new ApiException((int)res.StatusCode, err?["code"]?.GetValue<string>(),
err?["message"]?.GetValue<string>() ?? res.ReasonPhrase ?? "error");
}
public static async Task<JsonNode?> Call(string method, string path, object? body = null, IDictionary<string, string>? headers = null)
{
using var req = Request(method, path, body, headers);
using var res = await Http.SendAsync(req);
var text = await res.Content.ReadAsStringAsync();
JsonNode? env = null;
try { env = JsonNode.Parse(text); } catch (JsonException) { }
if (!res.IsSuccessStatusCode) throw ErrorFrom(res, env);
return env?["data"];
}
}
Step 2 — a token
Two kinds work:
-
Personal — needed for a run. Both lanes are metered in credits, so a run
needs a signed-in user (the page says “The audit run is metered, so it needs an
account.”, or the same of the instrument run) unless the estimate says
sponsor_enabled. Sign in and copy your token from /tokens.html, then export it asSKILLSAFE_TOKEN(the samples fall back to the placeholderYOUR_TOKEN). -
Guest — enough for
/meand/estimate.POST /guestwith{"slug": "pyobs-desk"}and noAuthorizationheader, exactly assdk.jsguest()does. The response’sdatacarriestokenandguest_id. Reuse a token you already hold rather than minting a new one each time.
# personal token (needed for a run): copy it from /tokens.html, then
# export SKILLSAFE_TOKEN=... and skip the rest of this block
# guest token (enough for /me and /estimate): no Authorization header on this one call
TOKEN=$(curl -sS -X POST "$BASE/guest" \
-H "Content-Type: application/json" \
-d '{"slug":"pyobs-desk"}' | jq -r '.data.token')
echo "guest token: ${TOKEN:0:12}..."
# personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if TOKEN in ("", "YOUR_TOKEN"):
TOKEN = "" # no Authorization header on /guest
guest = api("POST", "/guest", {"slug": "pyobs-desk"})
TOKEN = guest["token"]
print("guest", guest.get("guest_id"))
// personal token (needed for a run): paste it from /tokens.html into TOKEN and skip this
if (!TOKEN || TOKEN === "YOUR_TOKEN") {
TOKEN = ""; // no Authorization header on /guest
const guest = await api("POST", "/guest", { slug: "pyobs-desk" });
TOKEN = guest.token;
console.log("guest", guest.guest_id);
}
// inside main -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if token == "" {
var guest struct {
Token string `json:"token"`
GuestID string `json:"guest_id"`
}
if err := api("POST", "/guest", map[string]string{"slug": "pyobs-desk"}, nil, &guest); err != nil {
panic(err)
}
token = guest.Token
fmt.Println("guest", guest.GuestID)
}
// inside main -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if (token.isEmpty() || token.equals("YOUR_TOKEN")) {
token = ""; // no Authorization header on /guest
JsonNode guest = api("POST", "/guest", Map.of("slug", "pyobs-desk"), null);
token = guest.path("token").asText();
System.out.println("guest " + guest.path("guest_id").asText());
}
# personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if $token.empty? || $token == "YOUR_TOKEN"
$token = "" # no Authorization header on /guest
guest = api("POST", "/guest", { slug: "pyobs-desk" })
$token = guest["token"]
puts "guest #{guest["guest_id"]}"
end
// personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if ($TOKEN === "" || $TOKEN === "YOUR_TOKEN") {
$TOKEN = ""; // no Authorization header on /guest
$guest = api("POST", "/guest", ["slug" => "pyobs-desk"]);
$TOKEN = $guest->token;
echo "guest {$guest->guest_id}\n";
}
// top-level statements (above the Api class) -- personal token (needed for a run): set SKILLSAFE_TOKEN (from /tokens.html) and skip this
if (Api.Token == "" || Api.Token == "YOUR_TOKEN")
{
Api.Token = ""; // no Authorization header on /guest
var guest = await Api.Call("POST", "/guest", new { slug = "pyobs-desk" });
Api.Token = guest!["token"]!.GetValue<string>();
Console.WriteLine($"guest {guest["guest_id"]}");
}
Step 3 — GET /me
Who the token belongs to and what it can spend. sdk.js documents the shape as
{subject_type, subject_id, credits, profile?}. The page treats the caller as signed
in only when subject_type is "user", shows credits as the
balance (1 credit = $0.0001, so it divides by 10 000 for the dollar figure), and compares it
with the estimate in step 4 before it lets you run.
api GET /me | jq '.data // .error'
me = api("GET", "/me")
print(me["subject_type"], me.get("subject_id"), me.get("credits"))
signed_in = me["subject_type"] == "user"
const me = await api("GET", "/me");
console.log(me.subject_type, me.subject_id, me.credits);
const signedIn = me.subject_type === "user";
var me struct {
SubjectType string `json:"subject_type"`
SubjectID any `json:"subject_id"`
Credits any `json:"credits"`
}
if err := api("GET", "/me", nil, nil, &me); err != nil {
panic(err)
}
fmt.Println(me.SubjectType, me.SubjectID, me.Credits)
signedIn := me.SubjectType == "user"
_ = signedIn
JsonNode me = api("GET", "/me", null, null);
System.out.println(me.path("subject_type").asText() + " " + me.path("subject_id").asText() + " " + me.path("credits"));
boolean signedIn = me.path("subject_type").asText().equals("user");
me = api("GET", "/me")
puts [me["subject_type"], me["subject_id"], me["credits"]].join(" ")
signed_in = me["subject_type"] == "user"
$me = api("GET", "/me");
echo $me->subject_type, " ", $me->subject_id ?? "", " ", $me->credits ?? "", "\n";
$signedIn = $me->subject_type === "user";
var me = await Api.Call("GET", "/me");
Console.WriteLine($"{me?["subject_type"]} {me?["subject_id"]} {me?["credits"]}");
var signedIn = me?["subject_type"]?.GetValue<string>() == "user";
Step 4 — build the input, then POST /estimate
First a small makeInput that does what buildInput does: task,
then code as the paste the page would send, facts as a compact JSON string,
the note when it is not blank, and — for the instrument lane only — the
audit_handoff string when the file carries one. It reads the file
build-check.js (or handoff.js) printed. The samples build the
audit input of the worker example from worker-check.json; step 6 reuses
makeInput for the quotes (instrument) example. Then price it. The body of
/estimate is the same input you will send to /run; no job is created and
nothing is charged — /estimate is free. From the response the page reads:
| Field | Meaning |
|---|---|
hold_credits | A reservation, not a price: the worst case the platform holds while the run is in flight. The page shows it as “reserves N credits ($...) for the audit run” and adds that you are charged only for what the run actually uses, usually far less. What you pay is the job’s charged_credits after settlement (step 5). A balance below the hold is allowed, but the page warns that the reply may be cut short (truncated: true). |
min_credits | The smallest balance that can start a run. Below it the page disables the run button and names the shortfall, because the call would be a 402. |
sponsor_enabled | When true, a caller who is not signed in may run; otherwise the page asks them to sign in. |
warnings | Input warnings. They do not refuse the call; the page shows them next to the price (“input warnings: ...”), so check them before you run. |
The page re-prices whenever the paste, the note or the lane change. The amounts grow with the size
of code and facts, so read them from the call rather than hard-coding them.
# worker-check.json: printed by build-check.js (see "The facts string").
# (.facts | tojson) turns the facts into ONE compact string, as JSON.stringify does.
BODY=$(jq -c --arg note "Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?" \
'{task: "audit", code, facts: (.facts | tojson), note: $note}' worker-check.json)
# No page check at hand? facts: "{}" is accepted as a string, but the reply then has no page flags to answer.
# Instrument lane after an audit: read handoff.js's output instead and add its string:
# '{task: "instrument", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)'
api POST /estimate -d "$BODY" \
| jq '.error // (.data | {hold_credits, min_credits, sponsor_enabled, warnings})'
def make_input(task, check_path, note=None):
"""The body the page sends: task, code, facts as a JSON STRING, an optional note, and
(instrument lane only) the audit_handoff string -- every value a string."""
with open(check_path, encoding="utf-8") as f:
check = json.load(f) # build-check.js (or handoff.js) output
inp = {
"task": task,
"code": check["code"], # the paste as the page sends it
"facts": json.dumps(check["facts"], separators=(",", ":"), ensure_ascii=False), # a STRING, like JSON.stringify
}
n = (note or "").strip()
if n:
inp["note"] = n[:2000] # optional; the page caps it at 2000 chars
if task == "instrument" and check.get("audit_handoff"):
inp["audit_handoff"] = check["audit_handoff"] # optional; already a JSON string
return inp
inp = make_input("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?")
est = api("POST", "/estimate", inp)
print("hold", est.get("hold_credits"), "min", est.get("min_credits"),
"sponsored", est.get("sponsor_enabled"), "warnings", est.get("warnings") or [])
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (instrument lane only) the audit_handoff string -- every value a string.
function makeInput(task, checkPath, note) {
const check = JSON.parse(readFileSync(checkPath, "utf8")); // build-check.js (or handoff.js) output
const input = {
task,
code: check.code, // the paste as the page sends it
facts: JSON.stringify(check.facts), // a STRING: the page's free check
};
const n = (note || "").trim();
if (n) input.note = n.length > 2000 ? n.slice(0, 2000) : n; // optional; max 2000 chars
if (task === "instrument" && check.audit_handoff) input.audit_handoff = check.audit_handoff; // optional
return input;
}
const input = makeInput("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?");
const est = await api("POST", "/estimate", input);
console.log("hold", est.hold_credits, "min", est.min_credits,
"sponsored", est.sponsor_enabled, "warnings", est.warnings || []);
// makeInput builds the body the page sends: task, code, facts as a JSON STRING,
// an optional note and (instrument lane only) the audit_handoff string.
makeInput := func(task, checkPath, note string) map[string]any {
raw, err := os.ReadFile(checkPath) // build-check.js (or handoff.js) output
if err != nil {
panic(err)
}
var check struct {
Code string `json:"code"`
Facts json.RawMessage `json:"facts"`
AuditHandoff string `json:"audit_handoff"`
}
if err := json.Unmarshal(raw, &check); err != nil {
panic(err)
}
var facts bytes.Buffer
if err := json.Compact(&facts, check.Facts); err != nil { // keeps key order; travels as a STRING
panic(err)
}
in := map[string]any{"task": task, "code": check.Code, "facts": facts.String()}
if n := []rune(strings.TrimSpace(note)); len(n) > 0 {
if len(n) > 2000 {
n = n[:2000] // optional; the page caps it at 2000 chars
}
in["note"] = string(n)
}
if task == "instrument" && check.AuditHandoff != "" {
in["audit_handoff"] = check.AuditHandoff // optional; already a JSON string
}
return in
}
input := makeInput("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?")
var est struct {
HoldCredits any `json:"hold_credits"`
MinCredits any `json:"min_credits"`
SponsorEnabled bool `json:"sponsor_enabled"`
Warnings []any `json:"warnings"`
}
if err := api("POST", "/estimate", input, nil, &est); err != nil {
panic(err)
}
fmt.Printf("hold %v min %v sponsored %v warnings %v\n", est.HoldCredits, est.MinCredits, est.SponsorEnabled, est.Warnings)
// ---- add to the PyObsDesk class, next to api(): the body the page sends
static Map<String, Object> makeInput(String task, String checkPath, String note) throws Exception {
JsonNode check = JSON.readTree(Files.readString(Path.of(checkPath))); // build-check.js (or handoff.js) output
Map<String, Object> in = new LinkedHashMap<>();
in.put("task", task);
in.put("code", check.path("code").asText()); // the paste as the page sends it
in.put("facts", JSON.writeValueAsString(check.path("facts"))); // a STRING, compact
String n = note == null ? "" : note.strip();
if (!n.isEmpty()) in.put("note", n.length() > 2000 ? n.substring(0, 2000) : n); // optional; max 2000 chars
if (task.equals("instrument") && check.hasNonNull("audit_handoff")) in.put("audit_handoff", check.get("audit_handoff").asText());
return in;
}
// ---- inside main
Map<String, Object> input = makeInput("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?");
JsonNode est = api("POST", "/estimate", input, null);
System.out.println("hold " + est.path("hold_credits") + " min " + est.path("min_credits")
+ " sponsored " + est.path("sponsor_enabled").asBoolean(false) + " warnings " + est.path("warnings"));
# The body the page sends: task, code, facts as a JSON STRING, an optional note,
# and (instrument lane only) the audit_handoff string -- every value a string.
def make_input(task, check_path, note = nil)
check = JSON.parse(File.read(check_path, encoding: "UTF-8")) # build-check.js (or handoff.js) output
input = {
task: task,
code: check["code"], # the paste as the page sends it
facts: JSON.generate(check["facts"]) # a STRING, compact
}
n = note.to_s.strip
input[:note] = n[0, 2000] unless n.empty? # optional; max 2000 chars
input[:audit_handoff] = check["audit_handoff"] if task == "instrument" && !check["audit_handoff"].to_s.empty?
input
end
input = make_input("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?")
est = api("POST", "/estimate", input)
puts "hold #{est["hold_credits"]} min #{est["min_credits"]} " \
"sponsored #{est["sponsor_enabled"]} warnings #{(est["warnings"] || []).inspect}"
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (instrument lane only) the audit_handoff string -- every value a string.
function make_input(string $task, string $checkPath, string $note = ""): array {
// decode to objects (not arrays) so an empty {} stays {} when re-encoded
$check = json_decode(file_get_contents($checkPath)); // build-check.js (or handoff.js) output
$input = [
"task" => $task,
"code" => $check->code, // the paste as the page sends it
"facts" => json_encode($check->facts, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE), // a STRING
];
$n = trim($note);
if ($n !== "") $input["note"] = mb_substr($n, 0, 2000); // optional; max 2000 chars
if ($task === "instrument" && !empty($check->audit_handoff)) $input["audit_handoff"] = $check->audit_handoff;
return $input;
}
$input = make_input("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?");
$est = api("POST", "/estimate", $input);
echo "hold ", $est->hold_credits ?? "-", " min ", $est->min_credits ?? "-",
" sponsored ", json_encode($est->sponsor_enabled ?? false),
" warnings ", json_encode($est->warnings ?? []), "\n";
// The body the page sends: task, code, facts as a JSON STRING, an optional note,
// and (instrument lane only) the audit_handoff string -- every value a string.
static Dictionary<string, string> MakeInput(string task, string checkPath, string? note = null)
{
var compact = new JsonSerializerOptions { Encoder = System.Text.Encodings.Web.JavaScriptEncoder.UnsafeRelaxedJsonEscaping };
var check = JsonNode.Parse(File.ReadAllText(checkPath))!; // build-check.js (or handoff.js) output
var input = new Dictionary<string, string>
{
["task"] = task,
["code"] = check["code"]!.GetValue<string>(), // the paste as the page sends it
["facts"] = check["facts"]!.ToJsonString(compact), // a STRING, compact
};
var n = (note ?? "").Trim();
if (n.Length > 0) input["note"] = n.Length > 2000 ? n[..2000] : n; // optional; max 2000 chars
var handoff = check["audit_handoff"]?.GetValue<string>();
if (task == "instrument" && !string.IsNullOrEmpty(handoff)) input["audit_handoff"] = handoff;
return input;
}
var input = MakeInput("audit", "worker-check.json",
"Some invoices never arrive and nobody can tell why from the logs or the traces. What are we blind to?");
var est = await Api.Call("POST", "/estimate", input);
Console.WriteLine($"hold {est?["hold_credits"]} min {est?["min_credits"]} " +
$"sponsored {est?["sponsor_enabled"]} warnings {est?["warnings"]?.ToJsonString() ?? "[]"}");
Step 5 — POST /run, then poll the job (audit)
Send the same input to /run with an Idempotency-Key header, using a
personal token. Its data carries a job_id. Poll
GET /jobs/{job_id} the way sdk.js waitForJob does: every
second until status is succeeded or failed, giving up on
the client side after 180 seconds. On the finished job, read:
output.output— the model’s reply as a string: the JSON object from the output contract of the lane you asked for. Parse it yourself (from the first{to the last}). The samples print the verdict and headline, each signal’s status, each finding’sid,file,line,severityand number of edits, then each flag answer. Keep the text:recheck.jsreads it.truncated—truewhen the reply was cut short because the balance ran out. The page then closes the cut-off JSON, shows what arrived as a partial result (“N of 9 sections recovered” for an audit, of 8 for an instrument run), and says it was cut short by the available balance.charged_credits— the real cost of the run, as opposed to the estimate’shold_creditsreservation. The page re-reads/meafterwards for the new balance.error— on afailedjob.
Idempotency key. Resending a request with the same key must not start a second
paid run, so derive the key from the input, not from the clock — and include the lane, so an
audit and an instrument run of the same paste never share a key. The page uses
pyobs-desk:<lane>:<inputHash>:a<attempt>, where
<lane> is audit or instrument,
<inputHash> is a 32-bit djb2 hash of JSON.stringify(input) in hex, a
dash, and the string’s length in hex (app.js inputHash, over the
input without retry_note), and <attempt> is 1
for a normal run and 2 for the automatic reformat retry (which also adds
retry_note). For the two worked examples that is
pyobs-desk:audit:d354df7f-1c9d:a1 and pyobs-desk:instrument:713237d2-1059:a1. The format is the page’s
convention, not a requirement: any stable, unique string does. The samples below use the lane and a
SHA-256 prefix.
The page itself uses /run-stream (step 6), not /run; both take the
same body and header.
KEY="pyobs-desk:audit:$(printf '%s' "$BODY" | shasum -a 256 | cut -c1-16):a1"
JOB=$(api POST /run -H "Idempotency-Key: $KEY" -d "$BODY" | jq -r '.data.job_id')
for i in $(seq 180); do # 1 s interval, 180 s client-side limit
J=$(api GET "/jobs/$JOB")
S=$(printf '%s' "$J" | jq -r '.data.status')
if [ "$S" = "succeeded" ] || [ "$S" = "failed" ]; then break; fi
sleep 1
done
printf '%s' "$J" | jq '.data | {status, truncated, charged_credits, error}'
printf '%s' "$J" | jq -r '.data.output.output' \
| jq '{verdict, headline, signals: [(.signals // [])[] | {signal, status}],
findings: [(.findings // [])[] | {id, file, line, severity, edits: ((.edits // []) | length)}],
flag_answers: [(.flag_answers // [])[] | {flag, status}]}'
body = json.dumps(inp)
key = "pyobs-desk:audit:" + hashlib.sha256(body.encode()).hexdigest()[:16] + ":a1"
job_id = api("POST", "/run", inp, {"Idempotency-Key": key})["job_id"]
deadline = time.time() + 180 # 1 s interval, 180 s client-side limit
while True:
job = api("GET", "/jobs/" + urllib.parse.quote(job_id, safe=""))
if job["status"] in ("succeeded", "failed"):
break
if time.time() > deadline:
raise TimeoutError("job timed out")
time.sleep(1)
if job["status"] == "failed":
raise SystemExit(f"run failed: {job.get('error')}")
print("truncated:", job.get("truncated") is True, "charged:", job.get("charged_credits"))
text = (job.get("output") or {}).get("output") or ""
reply = json.loads(text[text.index("{"): text.rindex("}") + 1])
print(reply["verdict"], "-", reply["headline"])
for s in reply.get("signals", []):
print("signal", s["signal"], s["status"])
for f in reply.get("findings", []):
print(f["id"], f["file"], f.get("line"), f["severity"], len(f.get("edits") or []), "edit(s)")
for a in reply.get("flag_answers", []):
print("answer", a["flag"], a["status"])
const key = "pyobs-desk:audit:" +
createHash("sha256").update(JSON.stringify(input)).digest("hex").slice(0, 16) + ":a1";
const { job_id } = await api("POST", "/run", input, { "Idempotency-Key": key });
let job;
const deadline = Date.now() + 180000; // 1 s interval, 180 s client-side limit
for (;;) {
job = await api("GET", "/jobs/" + encodeURIComponent(job_id));
if (job.status === "succeeded" || job.status === "failed") break;
if (Date.now() > deadline) throw new Error("job timed out");
await new Promise((r) => setTimeout(r, 1000));
}
if (job.status === "failed") throw new Error("run failed: " + JSON.stringify(job.error));
console.log("truncated:", job.truncated === true, "charged:", job.charged_credits);
const text = (job.output && job.output.output) || "";
const reply = JSON.parse(text.slice(text.indexOf("{"), text.lastIndexOf("}") + 1));
console.log(reply.verdict, "-", reply.headline);
for (const s of reply.signals || []) console.log("signal", s.signal, s.status);
for (const f of reply.findings || []) console.log(f.id, f.file, f.line, f.severity, (f.edits || []).length, "edit(s)");
for (const a of reply.flag_answers || []) console.log("answer", a.flag, a.status);
inputJSON, _ := json.Marshal(input)
sum := sha256.Sum256(inputJSON)
key := "pyobs-desk:audit:" + hex.EncodeToString(sum[:8]) + ":a1"
var started struct {
JobID string `json:"job_id"`
}
if err := api("POST", "/run", input, map[string]string{"Idempotency-Key": key}, &started); err != nil {
panic(err)
}
type jobT struct {
Status string `json:"status"`
Output *struct {
Output string `json:"output"`
} `json:"output"`
Truncated bool `json:"truncated"`
ChargedCredits any `json:"charged_credits"`
Error any `json:"error"`
}
var job jobT
deadline := time.Now().Add(180 * time.Second) // 1 s interval, 180 s client-side limit
for {
job = jobT{}
if err := api("GET", "/jobs/"+url.PathEscape(started.JobID), nil, nil, &job); err != nil {
panic(err)
}
if job.Status == "succeeded" || job.Status == "failed" {
break
}
if time.Now().After(deadline) {
panic("job timed out")
}
time.Sleep(time.Second)
}
if job.Status == "failed" {
panic(fmt.Sprintf("run failed: %v", job.Error))
}
fmt.Println("truncated:", job.Truncated, "charged:", job.ChargedCredits)
text := ""
if job.Output != nil {
text = job.Output.Output
}
var reply map[string]any
if a, b := strings.Index(text, "{"), strings.LastIndex(text, "}"); a >= 0 && b > a {
if err := json.Unmarshal([]byte(text[a:b+1]), &reply); err != nil {
panic(err)
}
}
fmt.Println(reply["verdict"], "-", reply["headline"])
signals, _ := reply["signals"].([]any)
for _, e := range signals {
s, _ := e.(map[string]any)
fmt.Println("signal", s["signal"], s["status"])
}
findings, _ := reply["findings"].([]any)
for _, e := range findings {
f, _ := e.(map[string]any)
edits, _ := f["edits"].([]any)
fmt.Println(f["id"], f["file"], f["line"], f["severity"], len(edits), "edit(s)")
}
answers, _ := reply["flag_answers"].([]any)
for _, e := range answers {
a, _ := e.(map[string]any)
fmt.Println("answer", a["flag"], a["status"])
}
byte[] digest = MessageDigest.getInstance("SHA-256").digest(JSON.writeValueAsBytes(input));
String key = "pyobs-desk:audit:" + HexFormat.of().formatHex(digest, 0, 8) + ":a1";
String jobId = api("POST", "/run", input, Map.of("Idempotency-Key", key)).path("job_id").asText();
JsonNode job;
long deadline = System.currentTimeMillis() + 180_000; // 1 s interval, 180 s client-side limit
while (true) {
job = api("GET", "/jobs/" + URLEncoder.encode(jobId, StandardCharsets.UTF_8), null, null);
String s = job.path("status").asText();
if (s.equals("succeeded") || s.equals("failed")) break;
if (System.currentTimeMillis() > deadline) throw new RuntimeException("job timed out");
Thread.sleep(1000);
}
if (job.path("status").asText().equals("failed")) throw new RuntimeException("run failed: " + job.path("error"));
System.out.println("truncated: " + job.path("truncated").asBoolean(false) + ", charged: " + job.path("charged_credits"));
String text = job.path("output").path("output").asText("");
JsonNode reply = JSON.readTree(text.substring(text.indexOf('{'), text.lastIndexOf('}') + 1));
System.out.println(reply.path("verdict").asText() + " - " + reply.path("headline").asText());
for (JsonNode s : reply.path("signals")) {
System.out.println("signal " + s.path("signal").asText() + " " + s.path("status").asText());
}
for (JsonNode f : reply.path("findings")) {
System.out.println(f.path("id").asText() + " " + f.path("file").asText() + " " + f.path("line") + " "
+ f.path("severity").asText() + " " + f.path("edits").size() + " edit(s)");
}
for (JsonNode a : reply.path("flag_answers")) {
System.out.println("answer " + a.path("flag").asText() + " " + a.path("status").asText());
}
key = "pyobs-desk:audit:#{Digest::SHA256.hexdigest(JSON.generate(input))[0, 16]}:a1"
job_id = api("POST", "/run", input, { "Idempotency-Key" => key })["job_id"]
deadline = Time.now + 180 # 1 s interval, 180 s client-side limit
job = nil
loop do
job = api("GET", "/jobs/#{URI.encode_www_form_component(job_id)}")
break if %w[succeeded failed].include?(job["status"])
raise "job timed out" if Time.now > deadline
sleep 1
end
raise "run failed: #{job["error"].inspect}" if job["status"] == "failed"
puts "truncated: #{job["truncated"] == true}, charged: #{job["charged_credits"]}"
text = (job["output"] || {})["output"].to_s
reply = JSON.parse(text[text.index("{")..text.rindex("}")])
puts "#{reply["verdict"]} - #{reply["headline"]}"
(reply["signals"] || []).each { |s| puts "signal #{s["signal"]} #{s["status"]}" }
(reply["findings"] || []).each { |f| puts "#{f["id"]} #{f["file"]} #{f["line"].inspect} #{f["severity"]} #{(f["edits"] || []).size} edit(s)" }
(reply["flag_answers"] || []).each { |a| puts "answer #{a["flag"]} #{a["status"]}" }
$key = "pyobs-desk:audit:" . substr(hash("sha256", json_encode($input)), 0, 16) . ":a1";
$jobId = api("POST", "/run", $input, ["Idempotency-Key" => $key])->job_id;
$deadline = time() + 180; // 1 s interval, 180 s client-side limit
while (true) {
$job = api("GET", "/jobs/" . rawurlencode($jobId));
if (in_array($job->status, ["succeeded", "failed"], true)) break;
if (time() > $deadline) throw new RuntimeException("job timed out");
sleep(1);
}
if ($job->status === "failed") throw new RuntimeException("run failed: " . json_encode($job->error ?? null));
echo "truncated: ", json_encode(($job->truncated ?? false) === true), ", charged: ", $job->charged_credits ?? "-", "\n";
$text = $job->output->output ?? "";
$reply = json_decode(substr($text, strpos($text, "{"), strrpos($text, "}") - strpos($text, "{") + 1));
echo $reply->verdict, " - ", $reply->headline, "\n";
foreach ($reply->signals ?? [] as $s) echo "signal ", $s->signal, " ", $s->status, "\n";
foreach ($reply->findings ?? [] as $f) echo $f->id, " ", $f->file, " ", json_encode($f->line ?? null), " ", $f->severity, " ", count($f->edits ?? []), " edit(s)\n";
foreach ($reply->flag_answers ?? [] as $a) echo "answer ", $a->flag, " ", $a->status, "\n";
var digest = System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(input)));
var key = "pyobs-desk:audit:" + Convert.ToHexString(digest, 0, 8).ToLowerInvariant() + ":a1";
var started = await Api.Call("POST", "/run", input, new Dictionary<string, string> { ["Idempotency-Key"] = key });
var jobId = started!["job_id"]!.GetValue<string>();
JsonNode job;
var deadline = DateTime.UtcNow.AddSeconds(180); // 1 s interval, 180 s client-side limit
while (true)
{
job = (await Api.Call("GET", "/jobs/" + Uri.EscapeDataString(jobId)))!;
var s = job["status"]?.GetValue<string>();
if (s == "succeeded" || s == "failed") break;
if (DateTime.UtcNow > deadline) throw new TimeoutException("job timed out");
await Task.Delay(1000);
}
if (job["status"]?.GetValue<string>() == "failed") throw new Exception($"run failed: {job["error"]?.ToJsonString()}");
var truncated = job["truncated"] is JsonValue t && t.TryGetValue(out bool tb) && tb;
Console.WriteLine($"truncated: {truncated}, charged: {job["charged_credits"]}");
var text = job["output"]?["output"]?.GetValue<string>() ?? "";
var reply = JsonNode.Parse(text[text.IndexOf('{')..(text.LastIndexOf('}') + 1)])!;
Console.WriteLine($"{reply["verdict"]} - {reply["headline"]}");
foreach (var s in reply["signals"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"signal {s?["signal"]} {s?["status"]}");
foreach (var f in reply["findings"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"{f?["id"]} {f?["file"]} {f?["line"]?.ToJsonString() ?? "null"} {f?["severity"]} {f?["edits"]?.AsArray().Count ?? 0} edit(s)");
foreach (var a in reply["flag_answers"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"answer {a?["flag"]} {a?["status"]}");
Step 6 — POST /run-stream (Server-Sent Events, instrument)
The alternative to step 5, and what the page uses: same kind of body, same
Idempotency-Key header, one request. Run one or the other for a given input, not
both. To show the second lane, the samples build the instrument input of the
quotes example (with its own key) and stream it; to stream the audit input instead, pass the
step 4 input and key. The response is text/event-stream. sdk.js splits
it into frames on a blank line, takes the event name from an event: line (default
message) and concatenates the data: lines, skips frames with no data or
with data that is not JSON, and handles these events:
| Event | Data | What the SDK does |
|---|---|---|
delta | {text} | Passes text to onDelta. |
job | job info | Passes it to onJob (the page uses it to advance its progress display). |
done | the final payload: {job_id, status, charged_credits, output} | Keeps it as the result. |
pending | a payload | Treated exactly like done. If it carries a job_id and no output, poll that job as in step 5. |
error | {message, code, job_id} | After the stream ends, raises an error with those fields. |
If the response is not text/event-stream, it is an ordinary JSON envelope:
the SDK returns its data (an idempotent replay returns this way) or raises the
error on a non-2xx status. Handle both shapes.
From the final payload the page reads output.output (falling back to the
concatenated delta text), truncated and charged_credits.
Browsers receive progress ticks rather than deltas from this endpoint, so the page relies on the
final payload; if the stream fails part-way, it tries to recover a partial reply from whatever
delta text arrived. The samples print the verdict and headline, each returned file’s path and
size, each signal (kind, name, type, labels and
where), each setup step, each open item and each question. To put the files into the
paste and re-check them the way the page’s “Put every file into the paste” does,
keep the reply text and run recheck.js with the lane instrument. To
instrument a service you audited, read handoff.js’s output instead of
quotes-check.json: makeInput then carries its audit_handoff.
# the instrument worked example: quotes-check.json from build-check.js
# (for an audited service, read handoff.js's output instead; the + (...) adds its audit_handoff)
IBODY=$(jq -c --arg note "Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API." \
'{task: "instrument", code, facts: (.facts | tojson), note: $note} + (if .audit_handoff then {audit_handoff} else {} end)' \
quotes-check.json)
IKEY="pyobs-desk:instrument:$(printf '%s' "$IBODY" | shasum -a 256 | cut -c1-16):a1"
# -N turns off buffering; the whole stream lands in $STREAM
STREAM=$(api POST /run-stream -N -H "Idempotency-Key: $IKEY" -d "$IBODY")
# frames look like:
# event: job
# data: {...}
#
# event: done
# data: {"job_id":"...","status":"...","charged_credits":...,"output":{"output":"{\"task\":\"instrument\",...}"}}
# the final reply text, from the done (or pending) frame:
REPLY=$(printf '%s\n' "$STREAM" \
| awk '/^event:/{ev=$2} /^data:/ && (ev=="done" || ev=="pending"){sub(/^data: */,""); print}' \
| jq -r '.output.output')
# an idempotent replay answers with a plain JSON envelope instead: REPLY=$(printf '%s' "$STREAM" | jq -r '.data.output.output')
printf '%s' "$REPLY" | jq '{verdict, headline,
files: [(.files // [])[] | {path, characters: (.content | length)}],
signals: [(.signals // [])[] | {kind, name, type, labels, where}], setup, open, questions}'
# the instrument worked example: quotes-check.json from build-check.js
# (for an audited service, pass handoff.js's output instead: make_input adds its audit_handoff)
ins_inp = make_input("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.")
ins_key = "pyobs-desk:instrument:" + hashlib.sha256(json.dumps(ins_inp).encode()).hexdigest()[:16] + ":a1"
h = {"Content-Type": "application/json", "Authorization": "Bearer " + TOKEN, "Idempotency-Key": ins_key}
done, failure, streamed = None, None, ""
with requests.post(BASE + "/run-stream", headers=h, data=json.dumps(ins_inp), stream=True) as r:
if "text/event-stream" not in r.headers.get("content-type", ""):
try:
env = r.json()
except ValueError:
env = {}
if not r.ok:
err = env.get("error") or {}
raise ApiError(r.status_code, err.get("code"), err.get("message") or r.reason)
done = env.get("data")
else:
event, data = "message", ""
for line in r.iter_lines(decode_unicode=True):
if line == "": # a blank line ends a frame
if data:
try:
payload = json.loads(data)
except ValueError:
payload = None
if event == "delta" and payload is not None:
streamed += payload.get("text") or ""
elif event == "job":
print("job", payload)
elif event in ("done", "pending"):
done = payload
elif event == "error":
failure = payload
event, data = "message", ""
elif line.startswith("event:"):
event = line[6:].strip()
elif line.startswith("data:"):
data += line[5:].strip()
if failure:
raise SystemExit(f"{failure.get('code')}: {failure.get('message')} (job {failure.get('job_id')})")
done = done or {}
full = (done.get("output") or {}).get("output") or streamed
print("truncated:", done.get("truncated") is True, "charged:", done.get("charged_credits"))
result = json.loads(full[full.index("{"): full.rindex("}") + 1])
print(result["verdict"], "-", result["headline"])
for f in result.get("files", []):
print("file", f["path"], len(f.get("content") or ""), "characters")
for s in result.get("signals", []):
print("signal", s["kind"], s["name"], s.get("type"), ",".join(s.get("labels") or []) or "-", "in", s.get("where"))
for step in result.get("setup", []):
print("setup", step)
for o in result.get("open", []):
print("open", o["item"], "-", o.get("reason"))
for q in result.get("questions", []):
print("question", q)
// the instrument worked example: quotes-check.json from build-check.js
// (for an audited service, pass handoff.js's output instead: makeInput adds its audit_handoff)
const insInput = makeInput("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.");
const insKey = "pyobs-desk:instrument:" +
createHash("sha256").update(JSON.stringify(insInput)).digest("hex").slice(0, 16) + ":a1";
const res = await fetch(BASE + "/run-stream", {
method: "POST",
headers: { "Content-Type": "application/json", Authorization: "Bearer " + TOKEN, "Idempotency-Key": insKey },
body: JSON.stringify(insInput),
});
let done = null, failure = null, streamed = "";
if (!(res.headers.get("content-type") || "").includes("text/event-stream")) {
const json = await res.json().catch(() => ({}));
if (!res.ok) throw new Error((json.error && json.error.message) || res.statusText);
done = json.data;
} else {
const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
for (;;) {
const chunk = await reader.read();
if (chunk.done) break;
buffer += decoder.decode(chunk.value, { stream: true });
let idx;
while ((idx = buffer.indexOf("\n\n")) >= 0) {
const frame = buffer.slice(0, idx);
buffer = buffer.slice(idx + 2);
let event = "message", dataStr = "";
for (const line of frame.split("\n")) {
if (line.startsWith("event:")) event = line.slice(6).trim();
else if (line.startsWith("data:")) dataStr += line.slice(5).trim();
}
if (!dataStr) continue;
let data;
try { data = JSON.parse(dataStr); } catch { continue; }
if (event === "delta") streamed += data.text || "";
else if (event === "job") console.log("job", data);
else if (event === "done" || event === "pending") done = data;
else if (event === "error") failure = data;
}
}
}
if (failure) throw new Error(`${failure.code}: ${failure.message} (job ${failure.job_id})`);
const full = (done && done.output && done.output.output) || streamed;
console.log("truncated:", !!(done && done.truncated), "charged:", done && done.charged_credits);
const result = JSON.parse(full.slice(full.indexOf("{"), full.lastIndexOf("}") + 1));
console.log(result.verdict, "-", result.headline);
for (const f of result.files || []) console.log("file", f.path, (f.content || "").length, "characters");
for (const s of result.signals || []) console.log("signal", s.kind, s.name, s.type, (s.labels || []).join(",") || "-", "in", s.where);
for (const step of result.setup || []) console.log("setup", step);
for (const o of result.open || []) console.log("open", o.item, "-", o.reason);
for (const q of result.questions || []) console.log("question", q);
// the instrument worked example: quotes-check.json from build-check.js
// (for an audited service, pass handoff.js's output instead: makeInput adds its audit_handoff)
insInput := makeInput("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.")
insJSON, _ := json.Marshal(insInput)
insSum := sha256.Sum256(insJSON)
insKey := "pyobs-desk:instrument:" + hex.EncodeToString(insSum[:8]) + ":a1"
sreq, err := newRequest("POST", "/run-stream", insInput, map[string]string{"Idempotency-Key": insKey})
if err != nil {
panic(err)
}
sres, err := http.DefaultClient.Do(sreq)
if err != nil {
panic(err)
}
defer sres.Body.Close()
var done, failure map[string]any
var streamed strings.Builder
if !strings.Contains(sres.Header.Get("Content-Type"), "text/event-stream") {
var env envelope
_ = json.NewDecoder(sres.Body).Decode(&env)
if sres.StatusCode < 200 || sres.StatusCode > 299 {
panic(errorFrom(sres, env))
}
_ = json.Unmarshal(env.Data, &done)
} else {
sc := bufio.NewScanner(sres.Body)
sc.Buffer(make([]byte, 64*1024), 16*1024*1024) // the done frame can be long
event, data := "message", ""
for sc.Scan() {
line := sc.Text()
switch {
case line == "": // a blank line ends a frame
if data != "" {
var p map[string]any
if json.Unmarshal([]byte(data), &p) == nil {
switch event {
case "delta":
if t, ok := p["text"].(string); ok {
streamed.WriteString(t)
}
case "job":
fmt.Println("job", p)
case "done", "pending":
done = p
case "error":
failure = p
}
}
}
event, data = "message", ""
case strings.HasPrefix(line, "event:"):
event = strings.TrimSpace(line[6:])
case strings.HasPrefix(line, "data:"):
data += strings.TrimSpace(line[5:])
}
}
}
if failure != nil {
panic(fmt.Sprintf("%v: %v (job %v)", failure["code"], failure["message"], failure["job_id"]))
}
full := streamed.String()
if out, ok := done["output"].(map[string]any); ok {
if s, ok := out["output"].(string); ok && s != "" {
full = s
}
}
fmt.Println("truncated:", done["truncated"] == true, "charged:", done["charged_credits"])
var result map[string]any
if a, b := strings.Index(full, "{"), strings.LastIndex(full, "}"); a >= 0 && b > a {
if err := json.Unmarshal([]byte(full[a:b+1]), &result); err != nil {
panic(err)
}
}
fmt.Println(result["verdict"], "-", result["headline"])
files, _ := result["files"].([]any)
for _, e := range files {
f, _ := e.(map[string]any)
c, _ := f["content"].(string)
fmt.Println("file", f["path"], len([]rune(c)), "characters")
}
sigs, _ := result["signals"].([]any)
for _, e := range sigs {
s, _ := e.(map[string]any)
var labels []string
if ls, ok := s["labels"].([]any); ok {
for _, l := range ls {
labels = append(labels, fmt.Sprint(l))
}
}
lab := strings.Join(labels, ",")
if lab == "" {
lab = "-"
}
fmt.Println("signal", s["kind"], s["name"], s["type"], lab, "in", s["where"])
}
setup, _ := result["setup"].([]any)
for _, st := range setup {
fmt.Println("setup", st)
}
open, _ := result["open"].([]any)
for _, e := range open {
o, _ := e.(map[string]any)
fmt.Println("open", o["item"], "-", o["reason"])
}
questions, _ := result["questions"].([]any)
for _, q := range questions {
fmt.Println("question", q)
}
// the instrument worked example: quotes-check.json from build-check.js
// (for an audited service, pass handoff.js's output instead: makeInput adds its audit_handoff)
Map<String, Object> insInput = makeInput("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.");
byte[] insDigest = MessageDigest.getInstance("SHA-256").digest(JSON.writeValueAsBytes(insInput));
String insKey = "pyobs-desk:instrument:" + HexFormat.of().formatHex(insDigest, 0, 8) + ":a1";
HttpRequest sreq = request("POST", "/run-stream", insInput, Map.of("Idempotency-Key", insKey)).build();
HttpResponse<Stream<String>> sres = HTTP.send(sreq, HttpResponse.BodyHandlers.ofLines());
JsonNode done = null, failure = null;
StringBuilder streamed = new StringBuilder();
if (!sres.headers().firstValue("content-type").orElse("").contains("text/event-stream")) {
JsonNode env = JSON.readTree(sres.body().collect(Collectors.joining("\n")));
if (env == null) env = JSON.createObjectNode();
if (sres.statusCode() / 100 != 2) throw errorFrom(sres.statusCode(), env);
done = env.path("data");
} else {
String event = "message";
StringBuilder data = new StringBuilder();
Iterator<String> it = sres.body().iterator();
while (it.hasNext()) {
String line = it.next();
if (line.isEmpty()) { // a blank line ends a frame
if (data.length() > 0) {
JsonNode p = null;
try { p = JSON.readTree(data.toString()); } catch (Exception ignored) { }
if (p != null) {
switch (event) {
case "delta": streamed.append(p.path("text").asText("")); break;
case "job": System.out.println("job " + p); break;
case "done": case "pending": done = p; break;
case "error": failure = p; break;
default: break;
}
}
}
event = "message";
data.setLength(0);
} else if (line.startsWith("event:")) event = line.substring(6).trim();
else if (line.startsWith("data:")) data.append(line.substring(5).trim());
}
}
if (failure != null) {
throw new RuntimeException(failure.path("code").asText() + ": " + failure.path("message").asText()
+ " (job " + failure.path("job_id").asText() + ")");
}
String full = done == null ? "" : done.path("output").path("output").asText("");
if (full.isEmpty()) full = streamed.toString();
System.out.println("truncated: " + (done != null && done.path("truncated").asBoolean(false))
+ ", charged: " + (done == null ? "-" : done.path("charged_credits").toString()));
JsonNode result = JSON.readTree(full.substring(full.indexOf('{'), full.lastIndexOf('}') + 1));
System.out.println(result.path("verdict").asText() + " - " + result.path("headline").asText());
for (JsonNode f : result.path("files")) {
System.out.println("file " + f.path("path").asText() + " " + f.path("content").asText("").codePointCount(0, f.path("content").asText("").length()) + " characters");
}
for (JsonNode s : result.path("signals")) {
List<String> labels = new ArrayList<>();
s.path("labels").forEach(l -> labels.add(l.asText()));
System.out.println("signal " + s.path("kind").asText() + " " + s.path("name").asText() + " " + s.path("type").asText()
+ " " + (labels.isEmpty() ? "-" : String.join(",", labels)) + " in " + s.path("where").asText());
}
for (JsonNode step : result.path("setup")) System.out.println("setup " + step.asText());
for (JsonNode o : result.path("open")) System.out.println("open " + o.path("item").asText() + " - " + o.path("reason").asText());
for (JsonNode q : result.path("questions")) System.out.println("question " + q.asText());
# the instrument worked example: quotes-check.json from build-check.js
# (for an audited service, pass handoff.js's output instead: make_input adds its audit_handoff)
ins_input = make_input("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.")
ins_key = "pyobs-desk:instrument:#{Digest::SHA256.hexdigest(JSON.generate(ins_input))[0, 16]}:a1"
uri, req = build_request("POST", "/run-stream", ins_input, { "Idempotency-Key" => ins_key })
done = failure = nil
streamed = +""
Net::HTTP.start(uri.host, uri.port, use_ssl: true, read_timeout: 300) do |http|
http.request(req) do |res|
unless res["content-type"].to_s.include?("text/event-stream")
env = (JSON.parse(res.body) rescue {})
unless res.is_a?(Net::HTTPSuccess)
raise ApiError.new(res.code.to_i, env.dig("error", "code"), env.dig("error", "message") || res.message)
end
done = env["data"]
next
end
buf = +""
res.read_body do |chunk|
buf << chunk
while (i = buf.index("\n\n"))
frame = buf.slice!(0, i + 2)
event = "message"
data = +""
frame.split("\n").each do |line|
if line.start_with?("event:") then event = line[6..].strip
elsif line.start_with?("data:") then data << line[5..].strip
end
end
next if data.empty?
payload = (JSON.parse(data) rescue nil)
next if payload.nil?
case event
when "delta" then streamed << payload["text"].to_s
when "job" then puts "job #{payload.inspect}"
when "done", "pending" then done = payload
when "error" then failure = payload
end
end
end
end
end
raise "#{failure["code"]}: #{failure["message"]} (job #{failure["job_id"]})" if failure
done ||= {}
full = (done["output"] || {})["output"].to_s
full = streamed if full.empty?
puts "truncated: #{done["truncated"] == true}, charged: #{done["charged_credits"]}"
result = JSON.parse(full[full.index("{")..full.rindex("}")])
puts "#{result["verdict"]} - #{result["headline"]}"
(result["files"] || []).each { |f| puts "file #{f["path"]} #{f["content"].to_s.length} characters" }
(result["signals"] || []).each do |s|
labels = (s["labels"] || []).join(",")
puts "signal #{s["kind"]} #{s["name"]} #{s["type"]} #{labels.empty? ? "-" : labels} in #{s["where"]}"
end
(result["setup"] || []).each { |step| puts "setup #{step}" }
(result["open"] || []).each { |o| puts "open #{o["item"]} - #{o["reason"]}" }
(result["questions"] || []).each { |q| puts "question #{q}" }
// the instrument worked example: quotes-check.json from build-check.js
// (for an audited service, pass handoff.js's output instead: make_input adds its audit_handoff)
$insInput = make_input("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.");
$insKey = "pyobs-desk:instrument:" . substr(hash("sha256", json_encode($insInput)), 0, 16) . ":a1";
$done = null; $failure = null; $streamed = ""; $buf = ""; $ctype = "";
$handle = function (string $event, string $data) use (&$done, &$failure, &$streamed) {
$p = json_decode($data);
if ($p === null) return; // not JSON: skipped, as sdk.js does
switch ($event) {
case "delta": $streamed .= $p->text ?? ""; break;
case "job": echo "job ", json_encode($p), "\n"; break;
case "done": case "pending": $done = $p; break;
case "error": $failure = $p; break;
}
};
$ch = curl_init(BASE . "/run-stream");
curl_setopt_array($ch, [
CURLOPT_POST => true,
CURLOPT_HTTPHEADER => headers_for(["Idempotency-Key" => $insKey]),
CURLOPT_POSTFIELDS => json_encode($insInput, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE),
CURLOPT_HEADERFUNCTION => function ($ch, $line) use (&$ctype) {
if (stripos($line, "content-type:") === 0) $ctype = strtolower(trim(substr($line, 13)));
return strlen($line);
},
CURLOPT_WRITEFUNCTION => function ($ch, $chunk) use (&$buf, &$ctype, $handle) {
$buf .= $chunk; // collects the response in memory
if (strpos($ctype, "text/event-stream") === false) return strlen($chunk); // plain JSON: parse at the end
while (($i = strpos($buf, "\n\n")) !== false) {
$frame = substr($buf, 0, $i);
$buf = substr($buf, $i + 2);
$event = "message"; $data = "";
foreach (explode("\n", $frame) as $line) {
if (strpos($line, "event:") === 0) $event = trim(substr($line, 6));
elseif (strpos($line, "data:") === 0) $data .= trim(substr($line, 5));
}
if ($data !== "") $handle($event, $data);
}
return strlen($chunk);
},
]);
if (curl_exec($ch) === false) throw new RuntimeException(curl_error($ch));
$status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
curl_close($ch);
if (strpos($ctype, "text/event-stream") === false) {
$env = json_decode($buf) ?? new stdClass();
if ($status < 200 || $status > 299) {
throw new ApiError($status, $env->error->code ?? null, $env->error->message ?? "HTTP $status");
}
$done = $env->data ?? null;
}
if ($failure) throw new RuntimeException(($failure->code ?? "") . ": " . ($failure->message ?? "") . " (job " . ($failure->job_id ?? "") . ")");
$full = $done->output->output ?? "";
if ($full === "") $full = $streamed;
echo "truncated: ", json_encode(($done->truncated ?? false) === true), ", charged: ", $done->charged_credits ?? "-", "\n";
$result = json_decode(substr($full, strpos($full, "{"), strrpos($full, "}") - strpos($full, "{") + 1));
echo $result->verdict, " - ", $result->headline, "\n";
foreach ($result->files ?? [] as $f) echo "file ", $f->path, " ", mb_strlen($f->content ?? ""), " characters\n";
foreach ($result->signals ?? [] as $s) {
$labels = implode(",", $s->labels ?? []);
echo "signal ", $s->kind, " ", $s->name, " ", $s->type ?? "", " ", $labels === "" ? "-" : $labels, " in ", $s->where ?? "", "\n";
}
foreach ($result->setup ?? [] as $step) echo "setup ", $step, "\n";
foreach ($result->open ?? [] as $o) echo "open ", $o->item, " - ", $o->reason ?? "", "\n";
foreach ($result->questions ?? [] as $q) echo "question ", $q, "\n";
// the instrument worked example: quotes-check.json from build-check.js
// (for an audited service, pass handoff.js's output instead: MakeInput adds its audit_handoff)
var insInput = MakeInput("instrument", "quotes-check.json",
"Instrument this for Prometheus and an OpenTelemetry collector (Tempo behind it). Nothing in this module is instrumented yet; we can add prometheus-client and the OpenTelemetry API.");
var insDigest = System.Security.Cryptography.SHA256.HashData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(insInput)));
var insKey = "pyobs-desk:instrument:" + Convert.ToHexString(insDigest, 0, 8).ToLowerInvariant() + ":a1";
using var sreq = Api.Request("POST", "/run-stream", insInput, new Dictionary<string, string> { ["Idempotency-Key"] = insKey });
using var sres = await Api.Http.SendAsync(sreq, HttpCompletionOption.ResponseHeadersRead);
JsonNode? done = null, failure = null;
var streamed = new StringBuilder();
if (sres.Content.Headers.ContentType?.MediaType != "text/event-stream")
{
JsonNode? env = null;
try { env = JsonNode.Parse(await sres.Content.ReadAsStringAsync()); } catch (JsonException) { }
if (!sres.IsSuccessStatusCode) throw Api.ErrorFrom(sres, env);
done = env?["data"];
}
else
{
using var reader = new StreamReader(await sres.Content.ReadAsStreamAsync());
var ev = "message";
var data = new StringBuilder();
string? line;
while ((line = await reader.ReadLineAsync()) != null)
{
if (line.Length == 0) // a blank line ends a frame
{
if (data.Length > 0)
{
JsonNode? p = null;
try { p = JsonNode.Parse(data.ToString()); } catch (JsonException) { }
if (p != null)
{
switch (ev)
{
case "delta": streamed.Append(p["text"]?.GetValue<string>() ?? ""); break;
case "job": Console.WriteLine($"job {p.ToJsonString()}"); break;
case "done": case "pending": done = p; break;
case "error": failure = p; break;
}
}
}
ev = "message";
data.Clear();
}
else if (line.StartsWith("event:")) ev = line[6..].Trim();
else if (line.StartsWith("data:")) data.Append(line[5..].Trim());
}
}
if (failure != null) throw new Exception($"{failure["code"]}: {failure["message"]} (job {failure["job_id"]})");
var full = done?["output"]?["output"]?.GetValue<string>();
if (string.IsNullOrEmpty(full)) full = streamed.ToString();
var cut = done?["truncated"] is JsonValue tv && tv.TryGetValue(out bool cutShort) && cutShort;
Console.WriteLine($"truncated: {cut}, charged: {done?["charged_credits"]}");
var result = JsonNode.Parse(full[full.IndexOf('{')..(full.LastIndexOf('}') + 1)])!;
Console.WriteLine($"{result["verdict"]} - {result["headline"]}");
foreach (var f in result["files"]?.AsArray() ?? new JsonArray())
Console.WriteLine($"file {f?["path"]} {(f?["content"]?.GetValue<string>() ?? "").Length} characters");
foreach (var s in result["signals"]?.AsArray() ?? new JsonArray())
{
var labels = string.Join(",", (s?["labels"]?.AsArray() ?? new JsonArray()).Select(l => l?.GetValue<string>()));
Console.WriteLine($"signal {s?["kind"]} {s?["name"]} {s?["type"]} {(labels == "" ? "-" : labels)} in {s?["where"]}");
}
foreach (var step in result["setup"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"setup {step}");
foreach (var o in result["open"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"open {o?["item"]} - {o?["reason"]}");
foreach (var q in result["questions"]?.AsArray() ?? new JsonArray()) Console.WriteLine($"question {q}");
What the page does that you now do
- Reads the paste in the browser and builds
codeandfactsfrom it (obskit.js), joining dropped files under# file:lines and cutting a large paste on whole files with a# [pyobs-desk: not sent: ...]line; you send the same two strings. - Prices every input with
/estimateand refuses to run belowmin_credits, or on a paste with no Python it can read. - Sends each attempt with an input- and lane-derived
Idempotency-Key, and retries once withretry_notewhen a reply does not parse. - Closes a cut-off JSON reply and shows what arrived when
truncatedis true or the stream ends early. - Hands an audit to the instrument lane as
audit_handoff, with every edit applied to the paste first, and only while that edited paste is unchanged. - Checks the reply against the code (
recon.js), in the browser: files, lines, evidence, every edit applied in order and the free check re-run after each finding, flag answers and verdict; the instrumented files put back, re-checked and matched against the signals list; and repeated credentials.recheck.jsabove runs the same checks for you.
Or skip all of this and use PyObs Desk in the browser: the check, the flags, the handler, log, metric and span inventories, putting an edit or the instrumented files back into the paste, and the Markdown and CSV exports are free.