Jorijn van der Graaf
741e3a656d
The modules that turn a frozen range into passes and passes into a verdict: :Schema, :Tools (the DENY, git and base lists, a leaf, so the config and the pool both read them), :Plan, :Artifact, :Inventory, :Dispositions, :State, :Normalize, :Stream, :Prompt, :Config, and the URL half of :PgWire. Units, jobs, results, the config and the claude stream stay JSON from end to end, so every key keeps its place; :Text gives JSON values truthiness, get, ==, membership and Str(), and throws Dyn::Error under a name for each kind of unexpected data. Exactness where the bytes are seen: - the stream classifiers are hand-written scanners with case-insensitive folds (the Kelvin sign is a k), Unicode \b and \s, and `.` stopping at \n; the SQL call detection takes `<exe> sql` with any path before it; - config problems carry the JSON scanner's and the UTF-8 decoder's own texts, positions included; - a float total is Neumaier's compensated sum, and the build passes -ffp-contract=off: -march=native would fuse a*b+c; - `crafter-review inventory` and `config` print what the goldens record, byte for byte (the fixture and 14 configs, broken ones included); a message names this binary as the command to run. tests/Goldens: the recorded answers over the fixtures next to them (the commits fixture, a rich versioned series with CRLF, a folded subject, Markdown headings in and out of fences, the code-key rewrites); the suite compares 2304 results as DumpFile writes them, so a key out of order fails too. All agree. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
374 lines
19 KiB
C++
374 lines
19 KiB
C++
// SPDX-License-Identifier: AGPL-3.0-only
|
|
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
|
|
|
module Crafter.Review:Stream_impl;
|
|
import :Stream;
|
|
import :Json;
|
|
import :Text;
|
|
import :PgWire;
|
|
import std;
|
|
|
|
namespace Crafter::Review::Stream {
|
|
namespace {
|
|
constexpr std::size_t None = std::u32string_view::npos;
|
|
|
|
// An ASCII literal of a re.IGNORECASE pattern at `at`: a letter matches
|
|
// what re.I lets it match, anything else itself.
|
|
bool MatchAt(std::u32string_view text, std::size_t at, std::string_view literal) {
|
|
if (at + literal.size() > text.size()) return false;
|
|
for (std::size_t i = 0; i < literal.size(); ++i) {
|
|
const char p = literal[i];
|
|
const char32_t t = text[at + i];
|
|
if (p >= 'a' && p <= 'z') {
|
|
if (Dyn::IgnoreCaseLetter(t) != p) return false;
|
|
} else if (t != static_cast<char32_t>(static_cast<std::uint8_t>(p))) {
|
|
return false;
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
std::size_t FindCi(std::u32string_view text, std::string_view literal, std::size_t from, std::size_t to) {
|
|
if (to < literal.size()) return None;
|
|
for (std::size_t at = from; at + literal.size() <= to; ++at) {
|
|
if (MatchAt(text, at, literal)) return at;
|
|
}
|
|
return None;
|
|
}
|
|
|
|
bool ContainsCi(std::u32string_view text, std::string_view literal) {
|
|
return FindCi(text, literal, 0, text.size()) != None;
|
|
}
|
|
|
|
// `first.*(second|...)`: `.` stops at \n only, so both on one line. The
|
|
// first occurrence of `first` on a line leaves the most room after it.
|
|
bool ThenOnLine(std::u32string_view text, std::string_view first, std::initializer_list<std::string_view> seconds) {
|
|
std::size_t lineStart = 0;
|
|
while (lineStart <= text.size()) {
|
|
std::size_t lineEnd = text.find(U'\n', lineStart);
|
|
if (lineEnd == None) lineEnd = text.size();
|
|
const std::size_t at = FindCi(text, first, lineStart, lineEnd);
|
|
if (at != None) {
|
|
for (const std::string_view second : seconds) {
|
|
if (FindCi(text, second, at + first.size(), lineEnd) != None) return true;
|
|
}
|
|
}
|
|
if (lineEnd == text.size()) break;
|
|
lineStart = lineEnd + 1;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// \b<digits>\b
|
|
bool WordNumber(std::u32string_view text, std::u32string_view number) {
|
|
for (std::size_t at = text.find(number); at != None; at = text.find(number, at + 1)) {
|
|
const bool before = at == 0 || !Dyn::IsWord(text[at - 1]);
|
|
const bool after = at + number.size() == text.size() || !Dyn::IsWord(text[at + number.size()]);
|
|
if (before && after) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
std::size_t Spaces(std::u32string_view text, std::size_t at) {
|
|
std::size_t end = at;
|
|
while (end < text.size() && Dyn::IsSpace(text[end])) ++end;
|
|
return end - at;
|
|
}
|
|
|
|
// (^|[\s;&|(])\S*crafter-review\s+sql(\s|$): this binary's helper,
|
|
// spelled with any path before it.
|
|
bool HelperCall(std::u32string_view text) {
|
|
constexpr std::u32string_view Binary = U"crafter-review";
|
|
for (std::size_t at = text.find(Binary); at != None; at = text.find(Binary, at + 1)) {
|
|
const std::size_t gap = Spaces(text, at + Binary.size());
|
|
if (gap == 0) continue;
|
|
const std::size_t verb = at + Binary.size() + gap;
|
|
if (text.substr(verb, 3) != U"sql") continue;
|
|
if (verb + 3 == text.size() || Dyn::IsSpace(text[verb + 3])) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
// A tool_result's content as text: a string, or a list of text blocks.
|
|
std::string ContentText(const Json& content) {
|
|
if (content.is_string()) return content.get<std::string>();
|
|
if (!content.is_array()) return "";
|
|
std::vector<std::string> texts;
|
|
for (const Json& block : content) {
|
|
if (block.is_object() && Dyn::Get(block, "text").is_string()) texts.push_back(Dyn::At(block, "text").get<std::string>());
|
|
}
|
|
return Dyn::Join(texts, "\n");
|
|
}
|
|
|
|
// An object key as JSON writes it: a string as it is, null, true and
|
|
// false as themselves, a number as Dyn::ReprValue prints it.
|
|
std::string JsonKey(const Json& value) {
|
|
if (value.is_string()) return value.get<std::string>();
|
|
if (value.is_null()) return "null";
|
|
if (value.is_boolean()) return value.get<bool>() ? "true" : "false";
|
|
if (value.is_number()) return Dyn::ReprValue(value);
|
|
throw Dyn::Error("TypeError", std::format("unhashable type: '{}'", Dyn::TypeName(value)));
|
|
}
|
|
|
|
// A map key or set member, as text.
|
|
std::string KeyOf(const Json& value) {
|
|
if (value.is_array() || value.is_object()) throw Dyn::Error("TypeError", std::format("unhashable type: '{}'", Dyn::TypeName(value)));
|
|
return DumpCompactSorted(value);
|
|
}
|
|
|
|
const Json& GetOr(const Json& dict, std::string_view key, const Json& fallback) {
|
|
return Dyn::Has(dict, key) ? Dyn::At(dict, key) : fallback;
|
|
}
|
|
|
|
// `d.get(k, 0) or 0`
|
|
Json Tokens(const Json& usage, std::string_view key) {
|
|
const Json zero = 0;
|
|
const Json& value = GetOr(usage, key, zero);
|
|
return Dyn::Truthy(value) ? value : zero;
|
|
}
|
|
|
|
// The value as a float: ValueError for text that is no number,
|
|
// TypeError for anything that is neither.
|
|
double ToFloat(const Json& value) {
|
|
if (value.is_number_float()) return value.get<double>();
|
|
if (value.is_boolean()) return value.get<bool>() ? 1.0 : 0.0;
|
|
if (value.is_number_unsigned()) return static_cast<double>(value.get<std::uint64_t>());
|
|
if (value.is_number_integer()) return static_cast<double>(value.get<std::int64_t>());
|
|
if (value.is_string()) {
|
|
if (const std::optional<double> parsed = Dyn::ParseFloat(value.get_ref<const std::string&>())) return *parsed;
|
|
throw Dyn::Error("ValueError", Dyn::FloatError(value.get_ref<const std::string&>()));
|
|
}
|
|
throw Dyn::Error("TypeError", std::format("float() argument must be a string or a real number, not '{}'", Dyn::TypeName(value)));
|
|
}
|
|
}
|
|
|
|
bool IsSqlCall(std::string_view command) {
|
|
const std::u32string text = Dyn::Decode(command);
|
|
return HelperCall(text);
|
|
}
|
|
|
|
bool HasAnsweredLine(std::string_view text) {
|
|
for (std::size_t at = 0; at <= text.size();) {
|
|
if (text.substr(at).starts_with(PgWire::Answered)) return true;
|
|
const std::size_t newline = text.find('\n', at);
|
|
if (newline == std::string_view::npos) break;
|
|
at = newline + 1;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
bool LimitText(std::string_view text) {
|
|
const std::u32string t = Dyn::Decode(text);
|
|
return ThenOnLine(t, "out of", {"usage"}) || ContainsCi(t, "usage limit") || ContainsCi(t, "quota exceeded") || ContainsCi(t, "rate limit")
|
|
|| ContainsCi(t, "rate-limit") || ContainsCi(t, "session limit");
|
|
}
|
|
|
|
bool AuthText(std::string_view text) {
|
|
const std::u32string t = Dyn::Decode(text);
|
|
return WordNumber(t, U"401") || WordNumber(t, U"403") || ThenOnLine(t, "invalid", {"api key", "token", "credential"})
|
|
|| ContainsCi(t, "authentication_error") || ContainsCi(t, "authentication error") || ContainsCi(t, "unauthorised") || ContainsCi(t, "unauthorized")
|
|
|| ThenOnLine(t, "token", {"expired", "revoked"}) || ContainsCi(t, "not logged in") || ContainsCi(t, "/login");
|
|
}
|
|
|
|
bool CreditText(std::string_view text) {
|
|
const std::u32string t = Dyn::Decode(text);
|
|
// The last two: OpenRouter's 402.
|
|
constexpr std::array<std::string_view, 11> Phrases = {
|
|
"credit balance too low", "credit balance is too low", "add funds",
|
|
"reached your api usage limits", "reached your specified api usage limits", "reached your workspace api usage limits",
|
|
"reached your specified workspace api usage limits", "crossed its monthly api usage threshold", "enforced_spend_limit_reached",
|
|
"insufficient credits", "requires more credits",
|
|
};
|
|
return std::ranges::any_of(Phrases, [&](std::string_view phrase) { return ContainsCi(t, phrase); });
|
|
}
|
|
|
|
Json ParseStream(std::string_view text) {
|
|
Json result;
|
|
Json rate;
|
|
Json sessionId;
|
|
Json init;
|
|
Json usage = {{"input", 0}, {"cacheCreation", 0}, {"cacheRead", 0}, {"output", 0}};
|
|
Json first;
|
|
Json turns = 0;
|
|
Json toolCalls = Json::object();
|
|
std::int64_t toolErrors = 0;
|
|
std::map<std::string, Json> names;
|
|
std::set<std::string> sqlIds;
|
|
// The mcp__playwright__* calls and how many the server answered with
|
|
// an error; the SQL helper's calls and how many reached the database.
|
|
std::int64_t browserCalls = 0;
|
|
std::int64_t browserErrors = 0;
|
|
std::int64_t dbCalls = 0;
|
|
std::int64_t dbAnswered = 0;
|
|
for (const std::string& raw : Dyn::Split(text, "\n")) {
|
|
const std::string line = Dyn::Strip(raw);
|
|
if (!line.starts_with('{')) continue;
|
|
Json o;
|
|
try {
|
|
o = ParseLoose(line);
|
|
} catch (const std::exception&) {
|
|
continue;
|
|
}
|
|
if (!o.is_object()) continue;
|
|
const Json& type = Dyn::Get(o, "type");
|
|
if (!Dyn::Truthy(sessionId) && Dyn::Get(o, "session_id").is_string()) sessionId = Dyn::At(o, "session_id");
|
|
if (type == "system" && Dyn::Get(o, "subtype") == "init") {
|
|
init = o;
|
|
} else if (type == "result") {
|
|
result = o;
|
|
} else if (type == "rate_limit_event") {
|
|
const Json& info = Dyn::Get(o, "rate_limit_info");
|
|
if (Dyn::Truthy(info)) rate = info;
|
|
}
|
|
const Json& message = Dyn::Get(o, "message");
|
|
if (!message.is_object() || !Dyn::Truthy(message)) continue;
|
|
if (type == "assistant") {
|
|
const Json& u = Dyn::Or(Dyn::Get(message, "usage"), Dyn::EmptyDict());
|
|
if (Dyn::Truthy(u)) {
|
|
turns = Dyn::Add(turns, Json(1));
|
|
usage["input"] = Dyn::Add(usage["input"], Tokens(u, "input_tokens"));
|
|
usage["cacheCreation"] = Dyn::Add(usage["cacheCreation"], Tokens(u, "cache_creation_input_tokens"));
|
|
usage["cacheRead"] = Dyn::Add(usage["cacheRead"], Tokens(u, "cache_read_input_tokens"));
|
|
usage["output"] = Dyn::Add(usage["output"], Tokens(u, "output_tokens"));
|
|
if (first.is_null()) {
|
|
first = Json::object();
|
|
for (const std::string_view key : {"input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens", "output_tokens"}) {
|
|
first[std::string(key)] = Tokens(u, key);
|
|
}
|
|
}
|
|
}
|
|
for (const Json& block : Dyn::Or(Dyn::Get(message, "content"), Dyn::EmptyList())) {
|
|
if (!block.is_object() || Dyn::Get(block, "type") != "tool_use") continue;
|
|
const Json& name = Dyn::Get(block, "name");
|
|
const std::string key = JsonKey(name);
|
|
toolCalls[key] = Dyn::Add(Dyn::Has(toolCalls, key) ? toolCalls[key] : Json(0), Json(1));
|
|
names[KeyOf(Dyn::Get(block, "id"))] = name;
|
|
const Json& input = Dyn::Get(block, "input");
|
|
const Json command = input.is_object() ? Dyn::Get(Dyn::Or(input, Dyn::EmptyDict()), "command") : Json();
|
|
if (name == "Bash" && command.is_string() && IsSqlCall(command.get_ref<const std::string&>())) sqlIds.insert(KeyOf(Dyn::Get(block, "id")));
|
|
}
|
|
}
|
|
if (type == "user") {
|
|
for (const Json& block : Dyn::Or(Dyn::Get(message, "content"), Dyn::EmptyList())) {
|
|
if (!block.is_object() || Dyn::Get(block, "type") != "tool_result") continue;
|
|
const bool isError = Dyn::Truthy(Dyn::Get(block, "is_error"));
|
|
if (isError) ++toolErrors;
|
|
const std::string id = KeyOf(Dyn::Get(block, "tool_use_id"));
|
|
const auto named = names.find(id);
|
|
const std::string name = named == names.end() ? std::string() : Dyn::Str(named->second);
|
|
if (name.starts_with("mcp__playwright__")) {
|
|
++browserCalls;
|
|
if (isError) ++browserErrors;
|
|
}
|
|
if (sqlIds.contains(id)) {
|
|
++dbCalls;
|
|
if (HasAnsweredLine(ContentText(Dyn::Get(block, "content")))) ++dbAnswered;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Json cost;
|
|
Json denials = Json::array();
|
|
Json thinking = 0;
|
|
if (Dyn::Truthy(result) && Dyn::Get(result, "usage").is_object()) {
|
|
// The result line's totals are authoritative: stream-json repeats a
|
|
// turn's usage on every assistant content block, and its
|
|
// per-message output_tokens are placeholders (2026-09-17 smoke:
|
|
// summed 2.26M cache reads vs 985K real).
|
|
const Json& u = Dyn::At(result, "usage");
|
|
usage = {{"input", Tokens(u, "input_tokens")}, {"cacheCreation", Tokens(u, "cache_creation_input_tokens")},
|
|
{"cacheRead", Tokens(u, "cache_read_input_tokens")}, {"output", Tokens(u, "output_tokens")}};
|
|
const Json& details = Dyn::Or(Dyn::Get(u, "output_tokens_details"), Dyn::EmptyDict());
|
|
thinking = Tokens(details, "thinking_tokens");
|
|
cost = Dyn::Get(result, "total_cost_usd");
|
|
for (const Json& d : Dyn::Or(Dyn::Get(result, "permission_denials"), Dyn::EmptyList())) {
|
|
const Json& input = Dyn::Or(Dyn::Get(d, "tool_input"), Dyn::EmptyDict());
|
|
const Json& command = Dyn::Get(input, "command");
|
|
const Json& file = Dyn::Get(input, "file_path");
|
|
denials.push_back(Dyn::Truthy(command) ? command : (Dyn::Truthy(file) ? file : Dyn::Get(d, "tool_name")));
|
|
}
|
|
if (Dyn::IsInt(Dyn::Get(result, "num_turns"))) turns = Dyn::At(result, "num_turns");
|
|
}
|
|
usage["thinking"] = thinking;
|
|
return {
|
|
{"result", result},
|
|
{"rateLimit", rate},
|
|
{"sessionId", sessionId},
|
|
{"init", init},
|
|
{"usage", usage},
|
|
{"firstTurn", first},
|
|
{"turns", turns},
|
|
{"toolCalls", toolCalls},
|
|
{"toolErrors", toolErrors},
|
|
{"costUsd", cost},
|
|
{"denials", denials},
|
|
{"browser", {{"calls", browserCalls}, {"errors", browserErrors}}},
|
|
{"db", {{"calls", dbCalls}, {"answered", dbAnswered}}},
|
|
};
|
|
}
|
|
|
|
std::optional<Json> StructuredOutput(const Json& result) {
|
|
if (!Dyn::Truthy(result)) return std::nullopt;
|
|
Json inner = Dyn::Get(result, "structured_output");
|
|
const Json& text = Dyn::Get(result, "result");
|
|
if (inner.is_null() && text.is_string() && Dyn::Truthy(text)) {
|
|
try {
|
|
inner = ParseLoose(text.get_ref<const std::string&>());
|
|
} catch (const std::exception&) {
|
|
inner = nullptr;
|
|
}
|
|
}
|
|
if (inner.is_string()) {
|
|
try {
|
|
inner = ParseLoose(inner.get<std::string>());
|
|
} catch (const std::exception&) {
|
|
return std::nullopt;
|
|
}
|
|
}
|
|
if (!inner.is_object() || !Dyn::Get(inner, "findings").is_array()) return std::nullopt;
|
|
return inner;
|
|
}
|
|
|
|
std::pair<std::string, std::string> Classify(const Json& parsed, std::optional<std::int32_t> exitCode, std::string_view stderrText, bool timedOut) {
|
|
const Json& result = Dyn::At(parsed, "result");
|
|
const Json& rate = Dyn::Or(Dyn::At(parsed, "rateLimit"), Dyn::EmptyDict());
|
|
if (timedOut || exitCode == 124) return {"timeout", "process exceeded its time budget"};
|
|
// Asked only in its place in the `or`: whether it is asked at all is
|
|
// part of the classification.
|
|
auto rejected = [&] { return Dyn::Get(rate, "status") == "rejected"; };
|
|
if (!Dyn::Truthy(result)) {
|
|
if (CreditText(stderrText)) return {"credits", "credit balance exhausted (no result line)"};
|
|
if (rejected() || LimitText(stderrText)) return {"limit", "rate limit (no result line)"};
|
|
if (AuthText(stderrText)) return {"auth", "authentication failure (no result line)"};
|
|
return {"error", "no result line - the process did not finish"};
|
|
}
|
|
if (Dyn::Truthy(Dyn::Get(result, "is_error"))) {
|
|
std::vector<std::string> errors;
|
|
for (const Json& e : Dyn::Or(Dyn::Get(result, "errors"), Dyn::EmptyList())) errors.push_back(Dyn::Str(e));
|
|
const std::string message = std::format("{} {}", Dyn::StrOr(Dyn::Get(result, "result"), ""), Dyn::Join(errors, " "));
|
|
const std::string head = Dyn::Slice(message, 0, 200);
|
|
const Json& status = Dyn::Get(result, "api_error_status");
|
|
if (CreditText(message)) return {"credits", head};
|
|
if (Dyn::Equal(status, Json(401)) || Dyn::Equal(status, Json(403)) || AuthText(message)) return {"auth", head};
|
|
if (Dyn::Equal(status, Json(429)) || LimitText(message) || rejected()) return {"limit", head};
|
|
const Json& subtype = Dyn::Get(result, "subtype");
|
|
if (!Dyn::Truthy(subtype)) return {"error", std::format("error: {}", head)};
|
|
if (!subtype.is_string()) throw Dyn::Error("TypeError", std::format("unsupported operand type(s) for +: '{}' and 'str'", Dyn::TypeName(subtype)));
|
|
return {"error", std::format("{}: {}", subtype.get<std::string>(), head)};
|
|
}
|
|
if (!StructuredOutput(result)) return {"error", "no structured_output in the result envelope"};
|
|
return {"ok", ""};
|
|
}
|
|
|
|
Json Windows(const Json& rate) {
|
|
Json out = Json::object();
|
|
const Json& windows = Dyn::Or(Dyn::Get(Dyn::Or(rate, Dyn::EmptyDict()), "unifiedWindows"), Dyn::EmptyDict());
|
|
for (auto it = windows.begin(); it != windows.end(); ++it) {
|
|
const Json& w = it.value();
|
|
if (!w.is_object()) continue;
|
|
const Json& u = Dyn::Get(w, "utilization");
|
|
out[it.key()] = {{"utilization", u.is_null() ? Json() : Json(Dyn::Round(ToFloat(u) * 100.0, 1))}, {"resetsAt", Dyn::Get(w, "resetsAt")}};
|
|
}
|
|
return out;
|
|
}
|
|
}
|