feat(session): serve requests without blocking a thread, pause and step
The app's end of the connection, independent of the transport, which reaches it through Detail::Exchange: one request's way back, safe from any thread and never blocking. - What needs only the session's own state (hello, status, list, log, pause, resume) is answered on the transport thread. What touches the app (calls, get, set, trace, screenshot) queues for the next Pump and runs on the app's thread between frames. Waits and steps reply from the Pump that sees them satisfied, events from the thread that emits them. No request ever holds a thread while it waits. - A request whose timeout passes before Pump is answered Timeout and is guaranteed not to have run, so an agent never wonders whether a command went through; an Async result arrives on the Pump it is ready; requests left when the Session ends are answered ShuttingDown. - Advance() for frame loops and Paused()/TakeSteps() for fixed-step loops hold what the app gates on them; pausing an app that gates nothing is refused instead of silently doing nothing. pause --when stops the frame its condition holds; pause --on-check stops before the next Advance in the frame a check fails, optionally raising SIGTRAP for an attached debugger. - A wait that times out says what its condition's paths were; a log follower gets matching entries in order, from the backlog on, and is dropped when it cannot keep up. - Request parameters go through Codec, so a mistyped request gets the same did-you-mean replies a mistyped command does. The session sets PR_SET_PTRACER_ANY so a debugger the agent starts can attach. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
3500b73ecf
commit
bd64c44fc5
GPG key ID:
2937E59CDCC1BCFB
15 changed files with 1851 additions and 33 deletions
157
tests/ShouldWaitTraceAndFollow/main.cpp
Normal file
157
tests/ShouldWaitTraceAndFollow/main.cpp
Normal file
|
|
@ -0,0 +1,157 @@
|
|||
// SPDX-License-Identifier: LGPL-3.0-only
|
||||
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
||||
|
||||
// Waits, traces and log follows: `wait --frames` and a condition reply on the
|
||||
// Pump that satisfies them, a condition that cannot be decided replies with
|
||||
// why, an event wakes its waiter from the emitting thread without waiting
|
||||
// for Pump, and a timeout says what the condition's paths were. A trace
|
||||
// collects a value over frames with its stats, or a sample's history. The
|
||||
// log filters by level, and a follower receives new matching entries in
|
||||
// order, is dropped when it cannot keep up, and ends when its time is up.
|
||||
import std;
|
||||
import Crafter.Debug;
|
||||
using namespace Crafter::Debug;
|
||||
|
||||
namespace {
|
||||
std::int32_t failures = 0;
|
||||
|
||||
void Check(bool ok, std::string_view what) {
|
||||
std::println("{} {}", ok ? "PASS" : "FAIL", what);
|
||||
if (!ok) ++failures;
|
||||
}
|
||||
|
||||
class FakeExchange : public Detail::Exchange {
|
||||
public:
|
||||
std::atomic<bool> refuse = false;
|
||||
|
||||
void Reply(const Value& reply) override {
|
||||
std::lock_guard lock(mutex);
|
||||
if (!final) final = reply;
|
||||
}
|
||||
bool Stream(const Value& line) override {
|
||||
if (refuse) return false;
|
||||
std::lock_guard lock(mutex);
|
||||
lines.push_back(line);
|
||||
return true;
|
||||
}
|
||||
std::optional<Value> Final() {
|
||||
std::lock_guard lock(mutex);
|
||||
return final;
|
||||
}
|
||||
std::vector<Value> Lines() {
|
||||
std::lock_guard lock(mutex);
|
||||
return lines;
|
||||
}
|
||||
|
||||
private:
|
||||
std::mutex mutex;
|
||||
std::optional<Value> final;
|
||||
std::vector<Value> lines;
|
||||
};
|
||||
|
||||
Session* session = nullptr;
|
||||
|
||||
std::shared_ptr<FakeExchange> Send(std::string_view method, std::string_view params = "{}", std::int64_t timeoutMs = 30'000) {
|
||||
Value request = Value::MakeObject();
|
||||
request.Set("token", Value(session->Token()));
|
||||
request.Set("method", Value(method));
|
||||
request.Set("params", Value::Parse(params));
|
||||
request.Set("timeoutMs", Value(timeoutMs));
|
||||
auto exchange = std::make_shared<FakeExchange>();
|
||||
session->Handle(request, exchange);
|
||||
return exchange;
|
||||
}
|
||||
|
||||
Value ResultOf(FakeExchange& exchange) {
|
||||
const auto reply = exchange.Final();
|
||||
return reply && reply->Find("result") ? *reply->Find("result") : Value();
|
||||
}
|
||||
|
||||
std::string CodeOf(FakeExchange& exchange) {
|
||||
const auto reply = exchange.Final();
|
||||
return reply && reply->Find("error") ? reply->Find("error")->Find("code")->AsString() : std::string();
|
||||
}
|
||||
|
||||
std::string MessageOf(FakeExchange& exchange) {
|
||||
const auto reply = exchange.Final();
|
||||
return reply && reply->Find("error") ? reply->Find("error")->Find("message")->AsString() : std::string();
|
||||
}
|
||||
}
|
||||
|
||||
int main() {
|
||||
std::int32_t score = 0;
|
||||
Watch("score", score);
|
||||
Session owned({ .name = "wait", .listen = false });
|
||||
session = &owned;
|
||||
|
||||
auto frames = Send("debug.wait", R"({"frames": 3})");
|
||||
Pump();
|
||||
Pump();
|
||||
Check(!frames->Final(), "wait --frames 3 has not replied after two frames");
|
||||
Pump();
|
||||
Check(ResultOf(*frames).Find("frame")->AsInteger() == 3, "wait --frames 3 replies on the third");
|
||||
|
||||
auto condition = Send("debug.wait", R"({"condition": "score >= 2"})");
|
||||
for (std::int32_t i = 0; i < 5 && !condition->Final(); ++i) {
|
||||
++score;
|
||||
Pump();
|
||||
}
|
||||
Check(score == 2 && CodeOf(*condition).empty(), "a condition replies on the Pump it holds");
|
||||
auto broken = Send("debug.wait", R"({"condition": "missing > 1"})");
|
||||
Pump();
|
||||
Check(CodeOf(*broken) == "ConditionError" && MessageOf(*broken).find("no watch or sample 'missing'") != std::string::npos, "an undecidable condition says why");
|
||||
|
||||
auto event = Send("debug.wait", R"({"event": "Boom"})");
|
||||
std::thread([] { Emit("Boom", Value::Parse(R"({"power": 9})")); }).join();
|
||||
Check(event->Final().has_value() && ResultOf(*event).Find("payload")->Find("power")->AsInteger() == 9, "an event wakes its waiter without waiting for Pump");
|
||||
|
||||
auto slow = Send("debug.wait", R"({"condition": "score > 100"})", 1);
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
Pump();
|
||||
Check(CodeOf(*slow) == "Timeout" && MessageOf(*slow).find("score is 2") != std::string::npos, "a timeout reports the condition's paths");
|
||||
Check(CodeOf(*Send("debug.wait", R"({"frames": 1, "paused": true})")) == "InvalidParams", "waiting for two things at once is refused");
|
||||
|
||||
auto trace = Send("debug.trace", R"({"paths": ["score"], "frames": 4})");
|
||||
for (std::int32_t i = 0; i < 4; ++i) {
|
||||
score += 10;
|
||||
Pump();
|
||||
}
|
||||
const Value traced = ResultOf(*trace);
|
||||
const Value* series = traced.Find("series") ? traced.Find("series")->Find("score") : nullptr;
|
||||
Check(series != nullptr && series->Find("values")->Size() == 4 && series->Find("stats")->Find("min")->AsNumber() == 12
|
||||
&& series->Find("stats")->Find("max")->AsNumber() == 42 && series->Find("stats")->Find("changes")->AsInteger() == 3,
|
||||
"a trace collects a value per frame with its stats");
|
||||
|
||||
for (std::int32_t i = 1; i <= 3; ++i) {
|
||||
Sample("residual", 0.5 * i);
|
||||
Pump();
|
||||
}
|
||||
auto history = Send("debug.trace", R"({"paths": ["residual"], "history": true})");
|
||||
Pump();
|
||||
const Value historyReply = ResultOf(*history);
|
||||
const Value* samples = historyReply.Find("series") ? historyReply.Find("series")->Find("residual") : nullptr;
|
||||
Check(samples != nullptr && samples->Find("values")->Size() == 3 && samples->Find("stats")->Find("last")->AsNumber() == 1.5, "a trace can read a sample's history");
|
||||
|
||||
Log("net", "hello {}", 1);
|
||||
LogWarn("net", "careful");
|
||||
const Value warnings = ResultOf(*Send("debug.log", R"({"level": "warn"})"));
|
||||
Check(warnings.Find("entries")->Size() == 1 && warnings.Find("entries")->Items()[0].Find("message")->AsString() == "careful", "the log filters by level");
|
||||
Check(CodeOf(*Send("debug.log", R"({"level": "loud"})")) == "InvalidParams", "an unknown level is refused");
|
||||
|
||||
auto follow = Send("debug.log", R"({"follow": true, "category": "game"})");
|
||||
Log("game", "one");
|
||||
Log("other", "skip");
|
||||
Log("game", "two");
|
||||
const std::vector<Value> lines = follow->Lines();
|
||||
Check(lines.size() == 2 && lines[1].Find("entry")->Find("message")->AsString() == "two", "a follower receives new matching entries, in order");
|
||||
follow->refuse = true;
|
||||
Log("game", "three");
|
||||
Check(ResultOf(*Send("debug.status")).Find("following")->AsInteger() == 0, "a follower that cannot keep up is dropped");
|
||||
|
||||
auto brief = Send("debug.log", R"({"follow": true, "forMs": 1})");
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
Pump();
|
||||
Check(brief->Final().has_value() && ResultOf(*brief).Find("ended") != nullptr, "a follow ends when its time is up");
|
||||
|
||||
return failures == 0 ? 0 : 1;
|
||||
}
|
||||
Loading…
Reference in a new issue