Security review API

export LYBER_BASE_URL='https://lyber-paid-acceptance-gateway-production.up.railway.app/v1'
export LYBER_API_KEY='YOUR_LYBER_API_KEY'
export LYBER_MODEL='minimax-m3.1-flash-abliterated'

# Replace only this example's generated intake file.
rm -f intake.json
curl --max-time 600 --fail-with-body --silent --show-error \
  "$LYBER_BASE_URL/workflows/security-review" \
  -H "Authorization: Bearer $LYBER_API_KEY" \
  -H "Content-Type: application/json" \
  --data "{\"model\":\"$LYBER_MODEL\",\"stage\":\"intake\",\"messages\":[],\"max_tokens\":16384,\"max_cost_micro_usdc\":1000000}" \
  --output intake-pending.json &&
python3 - <<'PY'
import json
from pathlib import Path

pending = Path("intake-pending.json")
response = json.loads(pending.read_text(encoding="utf-8"))
reply = response["choices"][0]
message = reply["message"]
text = message.get("content")
if (reply.get("finish_reason") != "stop" or message.get("tool_calls")
        or message.get("role") != "assistant" or not isinstance(text, str) or not text.strip()
        or any(marker in text for marker in ("<tool_call", "<｜DSML｜", "[TOOL_CALL]", "<｜tool▁"))):
    raise SystemExit("Resolve the incomplete reply or tool call before continuing.")
pending.replace("intake.json")
print(message["content"])
PY

Audit type: API
Repository: [repository URL]
Revision: [frozen commit SHA and snapshot ID]
Scope: [routes or features to review]
Live server: [URL only if relevant; otherwise omit this line]

Repository tree:
[paste the relevant file tree]

--- path/to/route.ext @ [snapshot, actual lines] ---
[paste the complete relevant source]

--- path/to/auth.ext @ [snapshot, actual lines] ---
[paste authentication and authorization source]

--- path/to/service.ext @ [snapshot, actual lines] ---
[paste the internal calls, storage code and relevant configuration/tests]

if python3 - <<'PY'
import json
import os
from pathlib import Path

Path("scan-state.json").unlink(missing_ok=True)
reply = json.loads(Path("intake.json").read_text(encoding="utf-8"))["choices"][0]
message = reply["message"]
text = message.get("content")
if (reply.get("finish_reason") != "stop" or message.get("tool_calls")
        or message.get("role") != "assistant" or not isinstance(text, str) or not text.strip()
        or any(marker in text for marker in ("<tool_call", "<｜DSML｜", "[TOOL_CALL]", "<｜tool▁"))):
    raise SystemExit("Resolve the incomplete reply or tool call before continuing.")
context = Path("audit-context.txt").read_text(encoding="utf-8")
if not context.strip():
    raise SystemExit("Add your audit details and actual source to audit-context.txt.")
body = {
    "model": os.environ["LYBER_MODEL"], "stage": "scan",
    "messages": [message, {"role": "user", "content": context}],
    "max_tokens": 16384, "max_cost_micro_usdc": 1000000,
}
Path("scan-request-pending.json").write_text(json.dumps(body), encoding="utf-8")
PY
then
  curl --max-time 600 --fail-with-body --silent --show-error \
    "$LYBER_BASE_URL/workflows/security-review" \
    -H "Authorization: Bearer $LYBER_API_KEY" \
    -H "Content-Type: application/json" \
    --data-binary @scan-request-pending.json \
    --output scan-response-pending.json &&
  python3 - <<'PY'
import json
from pathlib import Path

request = json.loads(Path("scan-request-pending.json").read_text(encoding="utf-8"))
response = json.loads(Path("scan-response-pending.json").read_text(encoding="utf-8"))
choice = response["choices"][0]
message = choice["message"]
text = message.get("content")
if (choice.get("finish_reason") != "stop" or message.get("tool_calls")
        or message.get("role") != "assistant" or not isinstance(text, str) or not text.strip()
        or any(marker in text for marker in ("<tool_call", "<｜DSML｜", "[TOOL_CALL]", "<｜tool▁"))):
    raise SystemExit("Incomplete reply: inspect private pending files; no stage completed. Do not auto-replay.")
state = Path("scan-state-pending.json")
state.write_text(json.dumps({"request": request, "response": response}), encoding="utf-8")
state.replace("scan-state.json")
print(message["content"])
PY
else
  false
fi

export PREVIOUS_STAGE="scan"
export REVIEW_STAGE="discover"
export REVIEW_REPLY_FILE="review-packet.txt"
export REVIEW_CONTEXT_MODE="fresh" # Use history to keep full completed history.

if python3 - <<'PY'
import json
import os
from pathlib import Path

previous, stage = os.environ["PREVIOUS_STAGE"], os.environ["REVIEW_STAGE"]
allowed = {"scan", "discover", "deep_review", "recheck", "report"}
if previous not in allowed or stage not in allowed:
    raise SystemExit("Use a documented review stage.")
if stage != previous:
    Path(f"{stage}-state.json").unlink(missing_ok=True)
state = json.loads(Path(f"{previous}-state.json").read_text(encoding="utf-8"))
body, reply = state["request"], state["response"]["choices"][0]
message = reply["message"]
text = message.get("content")
if (reply.get("finish_reason") != "stop" or message.get("tool_calls")
        or message.get("role") != "assistant" or not isinstance(text, str) or not text.strip()
        or any(marker in text for marker in ("<tool_call", "<｜DSML｜", "[TOOL_CALL]", "<｜tool▁"))):
    raise SystemExit("Resolve the incomplete reply or tool call before continuing.")
reply_file = os.environ.get("REVIEW_REPLY_FILE")
if not reply_file:
    raise SystemExit("Set REVIEW_REPLY_FILE to your checked evidence packet or missing source.")
answer = Path(reply_file).read_text(encoding="utf-8")
if not answer.strip():
    raise SystemExit("The evidence packet must include actual source and counterevidence, not only IDs.")
mode = os.environ.get("REVIEW_CONTEXT_MODE", "fresh")
if mode not in {"fresh", "history"}:
    raise SystemExit("REVIEW_CONTEXT_MODE must be fresh or history.")
# Compact only after a completed stage. Never split an active tool loop.
base = [] if mode == "fresh" and stage != previous else [*body["messages"], message]
body["messages"] = [*base, {"role": "user", "content": answer}]
body["stage"] = stage
Path(f"{stage}-request-pending.json").write_text(json.dumps(body), encoding="utf-8")
PY
then
  curl --max-time 600 --fail-with-body --silent --show-error \
    "$LYBER_BASE_URL/workflows/security-review" \
    -H "Authorization: Bearer $LYBER_API_KEY" \
    -H "Content-Type: application/json" \
    --data-binary @"$REVIEW_STAGE-request-pending.json" \
    --output "$REVIEW_STAGE-response-pending.json" &&
  python3 - <<'PY'
import json
import os
from pathlib import Path

stage = os.environ["REVIEW_STAGE"]
request = json.loads(Path(f"{stage}-request-pending.json").read_text(encoding="utf-8"))
response = json.loads(Path(f"{stage}-response-pending.json").read_text(encoding="utf-8"))
choice = response["choices"][0]
message = choice["message"]
text = message.get("content")
if (choice.get("finish_reason") != "stop" or message.get("tool_calls")
        or message.get("role") != "assistant" or not isinstance(text, str) or not text.strip()
        or any(marker in text for marker in ("<tool_call", "<｜DSML｜", "[TOOL_CALL]", "<｜tool▁"))):
    raise SystemExit("Incomplete reply: inspect private pending files; no stage completed. Do not auto-replay.")
state = Path(f"{stage}-state-pending.json")
state.write_text(json.dumps({"request": request, "response": response}), encoding="utf-8")
state.replace(f"{stage}-state.json")
print(message["content"])
PY
else
  false
fi

// Run on your server. Never bundle the API key into a web page.
const base = process.env.LYBER_BASE_URL?.replace(/\/+$/, "");
const key = process.env.LYBER_API_KEY;
const model = process.env.LYBER_MODEL;
if (!base || !key || !model) throw new Error("Set the three environment variables first.");

const messages = []; // One private history per audit. Call sequentially.
// Provisional caller policy. Set these for your authorized workload.
const limits = { calls: 24, timeoutMs: 600000, costPerCall: 1000000 };
let calls = 0; // Each dispatch reserves one whole per-call ceiling.
const readFileTool = {
  type: "function", function: { name: "read_file", description: "Read a file in the frozen authorized source snapshot.",
    parameters: { type: "object", properties: { path: { type: "string" } }, required: ["path"], additionalProperties: false } },
};
const nativeSyntax = text => /<tool_call|<｜DSML｜|\[TOOL_CALL\]|<｜tool▁/.test(text ?? "");
function recovery(message, response, next) {
  const error = new Error(message); // Do not include raw source, provider errors or credentials.
  Object.defineProperties(error, {
    response: { value: response },
    requestMessages: { value: structuredClone(next) },
  });
  return error;
}
async function request(stage, next, tools) {
  if (calls >= limits.calls) throw recovery("unresolved: caller call/cost ceiling; retain missing evidence.", null, next);
  calls += 1; // Transport failures may still incur usage; never replay automatically.
  let response;
  try {
    response = await fetch(base + "/workflows/security-review", {
      method: "POST", redirect: "error", signal: AbortSignal.timeout(limits.timeoutMs),
      headers: { Authorization: "Bearer " + key, "Content-Type": "application/json" },
      body: JSON.stringify({ model, stage, messages: next, max_tokens: 16384,
        max_cost_micro_usdc: limits.costPerCall, ...(tools ? { tools } : {}) }),
    });
  } catch {
    throw recovery("Transport/deadline failure; usage uncertain. Inspect privately before another request.", null, next);
  }
  let raw;
  try { raw = await response.text(); }
  catch { throw recovery("Response transport failure; usage uncertain. Do not auto-replay.", null, next); }
  if (!response.ok) throw recovery("HTTP " + response.status + "; inspect private response before retrying.", raw, next);
  try { return JSON.parse(raw); }
  catch { throw recovery("Invalid JSON response; usage uncertain.", raw, next); }
}
async function review(stage, userText, { fresh = false, readFile } = {}) {
  if (fresh && (typeof userText !== "string" || !userText.trim())) throw new Error("Fresh stage requires a checked evidence packet.");
  let next = fresh ? [] : structuredClone(messages);
  if (userText !== undefined) next.push({ role: "user", content: userText });
  let invalidCalls = 0;
  const seenIds = new Set();
  for (let round = 0; round <= 8; round += 1) {
    const data = await request(stage, next, readFile ? [readFileTool] : undefined);
    const choice = data?.choices?.[0];
    const assistant = choice?.message;
    if (assistant?.role !== "assistant" || nativeSyntax(assistant.content)) {
      throw recovery("unresolved: malformed reply or leaked native tool syntax.", data, next);
    }
    if (choice.finish_reason === "tool_calls" && readFile && round < 8) {
      const tools = assistant.tool_calls;
      if (!Array.isArray(tools) || !tools.length || tools.some(call =>
          !call || typeof call !== "object" || typeof call.id !== "string" || !call.id || seenIds.has(call.id)) || new Set(tools.map(call => call.id)).size !== tools.length) {
        throw recovery("unresolved: invalid or duplicate tool IDs.", data, next);
      }
      next.push(assistant); // Preserve ALL fields, including required reasoning fields.
      let stop = false;
      for (const call of tools) {
        seenIds.add(call.id);
        let args;
        try { args = JSON.parse(call.function?.arguments); } catch { /* Return an actionable schema error below. */ }
        const valid = call.type === "function" && call.function?.name === "read_file"
          && args && typeof args === "object" && !Array.isArray(args)
          && typeof args.path === "string" && args.path.trim()
          && Object.keys(args).length === 1;
        let result;
        if (stop) result = { error: "unresolved: tool batch stopped after repeated malformed arguments; not executed." };
        else if (!valid) {
          invalidCalls += 1;
          result = { error: "read_file requires only nonempty string field path; example: {\"path\":\"src/routes.ts\"}. No arguments were inferred or executed." };
          stop = invalidCalls > 1;
        } else {
          try { result = { path: args.path, source: await readFile(args.path) }; }
          catch { result = { error: "Source read failed; supply the missing authorized frozen source." }; }
        }
        next.push({ role: "tool", tool_call_id: call.id, content: JSON.stringify(result) });
      }
      if (stop) throw recovery("unresolved: repeated malformed tool arguments; investigation stopped.", data, next);
      continue;
    }
    if (choice.finish_reason !== "stop" || assistant.tool_calls ||
        typeof assistant.content !== "string" || !assistant.content.trim()) {
      throw recovery("unresolved: incomplete output or tool round limit. Inspect partials; do not auto-replay.", data, next);
    }
    // Atomic publication. Human checks the stage artifact before starting the next stage.
    messages.splice(0, messages.length, ...next, assistant);
    return assistant.content;
  }
}

console.log(await review("intake"));
// Collect the user's answer and actual source, then:
// console.log(await review("scan", answerAndSource));
// Inspect each artifact, prepare a neutral packet with decisive source/counterevidence:
// console.log(await review("discover", discoveryPacket, { fresh: true }));
// console.log(await review("deep_review", neutralVerificationPacket, { fresh: true }));
// console.log(await review("recheck", fullFlowPacket, { fresh: true }));
// console.log(await review("report", adjudicatedPacket, { fresh: true }));
// Optional: { readFile: authorizedSnapshotReader }; the caller owns source access.