File size: 5,793 Bytes
3fd1a35 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | #include "ling3/flow_gate.h"
#include "ling3/chat_sampling.h"
#include <future>
#include <iostream>
#include <thread>
using namespace ling3::chat;
using namespace std::chrono_literals;
void Check(bool value, const char * message) { if (!value) throw std::runtime_error(message); }
template<class F> void Rejected(F f) {
bool rejected = false;
try { f(); } catch (const Error &) { rejected = true; }
Check(rejected, "stale or invalid command accepted");
}
int main() {
FlowGate gate;
gate.Begin("a", true);
Check(gate.Prepare() == 1, "sequence start");
auto waiting = std::async(std::launch::async, [&] { return gate.Wait([] { return true; }); });
Check(waiting.wait_for(30ms) == std::future_status::timeout, "generation advanced without ack");
Rejected([&] { gate.Ack("b", 1); });
Rejected([&] { gate.Ack("a", 2); });
gate.Ack("a", 1);
Check(waiting.get(), "ack did not resume");
gate.Ack("a", 1); // Idempotent retry cannot grant a future credit.
gate.Prepare();
Rejected([&] { gate.Ack("a", 1); });
auto canceled = std::async(std::launch::async, [&] { return gate.Wait([] { return true; }); });
Check(!gate.Cancel("b"), "canceled another request");
Check(canceled.wait_for(20ms) == std::future_status::timeout, "stale cancel unblocked generation");
Check(gate.Cancel("a") && !canceled.get(), "cancel while parked failed");
gate.End(); gate.Begin("b", true); gate.Prepare();
Rejected([&] { gate.Ack("a", 1); });
Check(!gate.Wait([] { return false; }), "disconnected parked client not canceled");
gate.End(); gate.Begin("parallel", false);
auto pause = std::async(std::launch::async, [&] { gate.Pause("parallel"); });
while (!gate.Status()["pause_requested"].get<bool>()) std::this_thread::yield();
auto paused = std::async(std::launch::async, [&] { return gate.Wait([] { return true; }); });
pause.get();
Check(gate.Status()["paused"], "pause returned before safe boundary");
Check(paused.wait_for(30ms) == std::future_status::timeout, "pause did not park generation");
Rejected([&] { gate.Resume("a"); });
gate.Resume("parallel"); Check(paused.get(), "resume failed"); gate.End();
Json body = {{"model", kModel}, {"messages", Json::array({{{"role", "user"}, {"content", "hi"}}})}};
for (auto bad : std::vector<Json>{{{"top_k", -1}}, {{"top_k", 1.5}}, {{"repeat_penalty", 0}},
{{"repeat_penalty", "1"}}, {{"enable_thinking", 1}}, {{"flow_control", "ack"}},
{{"reuse_generated_state", "true"}}}) {
auto request = body; request.update(bad); Rejected([&] { Parse(request); });
}
body.update({{"top_k", 2}, {"repeat_penalty", 2}, {"temperature", 0.7}, {"enable_thinking", true},
{"flow_control", "ack"}, {"stream", true}});
auto request = Parse(body);
Check(request.reuse_generated_state, "QA cache must be enabled by default");
auto incremental = body; incremental["reuse_generated_state"] = true;
Check(Parse(incremental).reuse_generated_state, "generated reuse flag missing");
gate.Begin("commit", false);
Check(gate.TryCommit(), "completion commit failed");
Check(!gate.Cancel("commit"), "late cancellation accepted after commit");
Check(!gate.TryCommit(), "duplicate commit accepted");
Rejected([&] { gate.Pause("commit"); });
gate.End();
gate.Begin("cancel-first", false);
Check(gate.Cancel("cancel-first") && !gate.TryCommit(), "canceled generation committed");
gate.End();
Check(request.prompt.ends_with("\n<think>"), "thinking template not enabled");
Check(request.prompt.find("detailed thinking on") != std::string::npos, "thinking instruction missing");
Check(request.flow_ack, "flow mode missing");
auto conflicting = body;
conflicting["enable_thinking"] = false;
conflicting["session_id"] = "conversation-a";
conflicting["messages"].insert(conflicting["messages"].begin(),
Json{{"role", "system"}, {"content", "detailed thinking on"}});
const auto normalized = Parse(conflicting);
Check(normalized.session_id == "conversation-a" && normalized.prompt.find("detailed thinking on") == std::string::npos
&& normalized.prompt.ends_with("<think></think>"), "thinking switch did not override system");
conflicting["enable_thinking"] = true;
conflicting["messages"][0]["content"] = "detailed thinking off; detailed thinking on";
const auto enabled = Parse(conflicting);
Check(enabled.prompt.find("detailed thinking off") == std::string::npos && enabled.prompt.ends_with("<think>"),
"thinking enable did not override mixed system instructions");
conflicting["session_id"] = ""; Rejected([&] { Parse(conflicting); });
std::mt19937 random(42);
request.temperature = 0;
Check(SampleToken(std::vector<float>{6, 4, -1}, request, {true, false, false}, random) == 1,
"positive repetition penalty not applied before greedy selection");
Check(SampleToken(std::vector<float>{-1, -1.5F, -3}, request, {true, false, false}, random) == 1,
"negative repetition penalty has wrong sign");
request.temperature = 1; request.repeat_penalty = 1; request.top_k = 2;
for (int i=0; i<500; ++i)
Check(SampleToken(std::vector<float>{3, 2, 1, 0}, request, {}, random) < 2, "top_k tail sampled");
request.top_k = 1;
Check(SampleToken(std::vector<float>{1, 2, 0}, request, {}, random) == 1, "top_k=1 not deterministic");
request.top_k = 0; request.top_p = 0.01;
Check(SampleToken(std::vector<float>{1, 2, 0}, request, {}, random) == 1, "top_p prefix ignored");
std::cout << "flow ack, stale controls, parked cancellation, pause/resume, sampling and thinking passed\n";
}
|