WHC/Hermit
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

feat: add persistent multi-model workers and orchestration budgets

Browse Source
This commit is contained in:
WHC
2026-10-19 11:29:26 +08:00
parent 235652604d
commit 143dba203b
14 changed files with 1043 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_subagent.c
+125 -1
View File
@@ -11,6 +11,7 @@
#include "cJSON.h"
#include "cst_agent.h"
#include "cst_subagent.h"
#include "cst_models.h"
#include "cst_tools.h"
#include "cst_util.h"
#include "os_compat.h"
@@ -42,6 +43,12 @@ static int nfail;
static atomic_int g_child_active, g_child_max, g_child_reqs;
static atomic_int g_child_saw_task, g_child_saw_write, g_child_role_prompt;
static atomic_int g_parent_saw_task;
static atomic_int g_worker_model_seen, g_worker_history_seen;
static atomic_int g_worker_key_seen;
static pthread_mutex_t cache_mu = PTHREAD_MUTEX_INITIALIZER;
static char worker_cache_key[128];
static atomic_int worker_cache_mismatch;
static atomic_int fail_next_worker_request;
static void send_all(test_sock_t fd, const char *s, size_t n) {
while (n > 0) {
@@ -161,11 +168,35 @@ static void handle(test_sock_t fd, const char *body_txt) {
cJSON_IsString(c))
user = c->valuestring;
}
int child = cJSON_IsString(sys) && strstr(sys->valuestring, "SUB-AGENT");
if (child && atomic_exchange(&fail_next_worker_request, 0)) {
atomic_fetch_add(&g_child_reqs, 1);
const char *failure = "HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
send_all(fd, failure, strlen(failure));
cJSON_Delete(body);
return;
}
const char *hdr = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n"
"Connection: close\r\n\r\n";
send_all(fd, hdr, strlen(hdr));
int child = cJSON_IsString(sys) && strstr(sys->valuestring, "SUB-AGENT");
if (child) {
const cJSON *model = cJSON_GetObjectItemCaseSensitive(body, "model");
if (cJSON_IsString(model) && strcmp(model->valuestring, "cheap-worker") == 0) {
atomic_store(&g_worker_model_seen, 1);
const cJSON *key = cJSON_GetObjectItemCaseSensitive(body, "prompt_cache_key");
pthread_mutex_lock(&cache_mu);
if (!cJSON_IsString(key)) atomic_store(&worker_cache_mismatch, 1);
else if (!worker_cache_key[0]) snprintf(worker_cache_key, sizeof(worker_cache_key), "%s", key->valuestring);
else if (strcmp(worker_cache_key, key->valuestring)) atomic_store(&worker_cache_mismatch, 1);
pthread_mutex_unlock(&cache_mu);
}
int assignments = 0;
for (int i = 0; i < nm; i++) {
const cJSON *m = cJSON_GetArrayItem(msgs, i);
const cJSON *role = cJSON_GetObjectItemCaseSensitive(m, "role");
if (cJSON_IsString(role) && strcmp(role->valuestring, "user") == 0) assignments++;
}
if (assignments >= 2) atomic_store(&g_worker_history_seen, 1);
atomic_fetch_add(&g_child_reqs, 1);
int now = atomic_fetch_add(&g_child_active, 1) + 1;
int mx = atomic_load(&g_child_max);
@@ -252,6 +283,8 @@ static void *conn_main(void *p) {
if (hdr_end && (long)(req.len - hdr_end) >= need)
break;
}
if (hdr_end && strstr(req.buf, "Authorization: Bearer worker-secret"))
atomic_store(&g_worker_key_seen, 1);
if (hdr_end)
handle(fd, req.buf + hdr_end);
cst_buf_free(&req);
@@ -489,7 +522,98 @@ int main(void) {
CHECK(nrec == 3 && strcmp(recs[0].status, "cancelled") == 0,
"cancelled run recorded");
/* Configuration validates bindings and never stores credentials. */
CHECK(cst_workers_config_set(ws, "{\"max_steps\":0}", err, sizeof(err)) != 0,
"reject zero budget");
CHECK(cst_workers_config_set(ws, "{\"max_steps\":1.5}", err, sizeof(err)) != 0,
"reject fractional budget");
CHECK(cst_workers_config_set(ws, "{\"roles\":{\"general\":\"missing\"}}", err, sizeof(err)) != 0,
"reject missing model profile");
char provider_doc[1024];
snprintf(provider_doc, sizeof(provider_doc),
"{\"providers\":{\"worker-test\":{\"api_base\":\"%s\",\"key_env\":\"CST_TEST_WORKER_KEY\",\"default_model\":\"cheap-worker\"}},\"models\":[{\"match\":\"cheap-worker\",\"cache\":\"openai\"}]}", cfg.api_base);
CHECK(cst_models_load_json(provider_doc, NULL, err, sizeof(err)) == 0,
"register independent worker provider");
setenv("CST_TEST_WORKER_KEY", "worker-secret", 1);
CHECK(cst_workers_config_set(ws,
"{\"max_steps\":20,\"max_tasks\":8,\"max_requests\":100,"
"\"profiles\":{\"cheap\":{\"provider\":\"worker-test\",\"model\":\"cheap-worker\"}},"
"\"roles\":{\"explore\":\"cheap\"}}", err, sizeof(err)) == 0,
"save bound worker profile");
/* Cancellation belongs to the previous turn; start with a fresh parent. */
cst_agent_destroy(a);
a = cst_agent_create(&cfg);
ctx.agent = a;
cst_workers_config_apply(cst_agent_config_mut(a));
cst_worker_budget_reset(cst_agent_config_mut(a)->worker_budget);
c.args[1].value = "Find C"; c.args[1].value_len = 6;
c.args[2].value = "explore"; c.args[2].value_len = 7;
memset(&r, 0, sizeof(r));
cst_tool_run(&ctx, &c, &r);
if (r.exit_code != 0) fprintf(stderr, "worker profile result: %s\n", r.out ? r.out : "empty");
CHECK(r.exit_code == 0 && r.out && strstr(r.out, "worker_id=sub-4") &&
atomic_load(&g_worker_model_seen) && atomic_load(&g_worker_key_seen),
"worker uses bound provider/model and its own credential");
cst_tool_result_free(&r);
/* Recreate the parent to prove continuation is disk-backed. */
cst_agent_destroy(a);
a = cst_agent_create(&cfg); ctx.agent = a;
c.nargs = 4;
snprintf(c.args[3].key, sizeof(c.args[3].key), "worker_id");
c.args[3].value = "sub-4"; c.args[3].value_len = 5;
c.args[1].value = "Find D";
memset(&r, 0, sizeof(r));
cst_tool_run(&ctx, &c, &r);
if (r.exit_code != 0) fprintf(stderr, "worker resume result: %s\n", r.out ? r.out : "empty");
CHECK(r.exit_code == 0 && atomic_load(&g_worker_history_seen) &&
r.out && strstr(r.out, "child report D"), "worker resumes after parent restart");
cst_tool_result_free(&r);
c.nargs = 3;
c.args[1].value = "Find F";
memset(&r, 0, sizeof(r));
cst_tool_run(&ctx, &c, &r);
CHECK(r.exit_code == 0 && worker_cache_key[0] && !atomic_load(&worker_cache_mismatch),
"same role and model share a stable cache route across worker IDs");
cst_tool_result_free(&r);
c.nargs = 4;
cst_worker_budget_reset(cst_agent_config_mut(a)->worker_budget);
cst_agent_config_mut(a)->subagent_total = 1;
c.args[1].value = "Find E";
memset(&r, 0, sizeof(r));
int before = atomic_load(&g_child_reqs);
cst_tool_run(&ctx, &c, &r);
CHECK(r.exit_code != 0 && atomic_load(&g_child_reqs) == before + 1,
"shared request budget stops before second HTTP request");
cst_tool_result_free(&r);
cst_worker_budget_reset(cst_agent_config_mut(a)->worker_budget);
atomic_store(&fail_next_worker_request, 1);
memset(&r, 0, sizeof(r));
before = atomic_load(&g_child_reqs);
cst_tool_run(&ctx, &c, &r);
CHECK(r.exit_code != 0 && atomic_load(&g_child_reqs) == before + 1 &&
r.out && strstr(r.out, "budget exhausted"),
"HTTP retry consumes budget and cannot bypass one-request cap");
cst_tool_result_free(&r);
cst_agent_config_mut(a)->subagent_tasks = 1;
memset(&r, 0, sizeof(r));
before = atomic_load(&g_child_reqs);
cst_tool_run(&ctx, &c, &r);
CHECK(r.exit_code != 0 && atomic_load(&g_child_reqs) == before,
"delegation cap rejects without HTTP request");
cst_tool_result_free(&r);
cst_agent_config_mut(a)->orchestrate = 1;
CHECK(!cst_tool_active(cst_agent_config(a), cst_tool_find("write_file")) &&
!cst_tool_active(cst_agent_config(a), cst_tool_find("run_shell")) &&
cst_tool_active(cst_agent_config(a), td), "coordinator delegates mutations");
CHECK(cst_workers_config_set(ws,
"{\"max_steps\":20,\"max_tasks\":8,\"max_requests\":3}", err, sizeof(err)) == 0,
"configure small concurrent budget");
before = atomic_load(&g_child_reqs);
rc = cst_agent_run_turn_ex(a, "concurrent budget test", &h, err, sizeof(err));
CHECK(atomic_load(&g_child_reqs) - before == 3,
"parallel children cannot exceed shared HTTP request cap");
cst_agent_destroy(a);
unsetenv("CST_TEST_WORKER_KEY");
cst_tool_unregister_owner("builtin:task");
CHECK(cst_agent_store_delete(ws, "parent1") == 0, "delete session");
CHECK(cst_subagent_list(ws, "parent1", recs, 8) == 0,