#ifndef _DEFAULT_SOURCE #define _DEFAULT_SOURCE #endif #ifndef _XOPEN_SOURCE #define _XOPEN_SOURCE 700 #endif #include "cst_subagent.h" #include #include #include #include #include #include #include #include "cJSON.h" #include "cst_agent.h" #include "cst_parliament.h" #include "cst_skills.h" #include "cst_tools.h" #include "cst_runtime.h" #include "cst_worktree.h" #include "cst_util.h" #include "os_compat.h" #define ROLE_FILE_MAX (64 * 1024) #define INDEX_MAX 2048 static int worker_active(const char *path); static void worker_live(const char *path, cst_subagent_rec_t *record); static pthread_mutex_t g_hook_mu = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t g_seq_mu = PTHREAD_MUTEX_INITIALIZER; static char g_ws[OS_PATH_MAX]; /* ---------- roles ---------- */ static const struct { const char *name, *desc, *prompt; int read_only; } BUILTIN[] = { {"general", "General-purpose agent for multi-step work: researches, edits files and " "runs commands as needed.", "", 0}, {"explore", "Fast read-only explorer: searches and reads the workspace to answer a " "question (where is X, how does Y work, which files do Z). Several can " "run at once.", "You only read. Search broadly first (grep, glob, list_dir), then read " "just the relevant parts. Report concrete findings: file paths with line " "numbers, short excerpts when they matter, and a direct answer. Say what " "you looked for and did not find.", 1}, {"plan", "Read-only planner: investigates the material involved and returns a " "step-by-step implementation plan.", "You only read. Investigate what is involved, then return a concrete " "plan: what to produce or change and where, risks and edge cases, and " "how to check the result. Sketch code only where it removes ambiguity.", 1}, }; #define NBUILTIN ((int)(sizeof(BUILTIN) / sizeof(BUILTIN[0]))) static char *slurp(const char *path, size_t max) { FILE *fp = fopen(path, "rb"); if (!fp) return NULL; cst_buf_t b = {0}; char tmp[8192]; size_t n; while ((n = fread(tmp, 1, sizeof(tmp), fp)) > 0) { if (b.len + n > max) { cst_buf_free(&b); fclose(fp); return NULL; } cst_buf_append(&b, tmp, n); } fclose(fp); if (!b.buf) cst_buf_puts(&b, ""); cst_strip_bom(b.buf, &b.len); return cst_buf_steal(&b); } static int save_atomic(const char *path, const char *text) { char tmp[OS_PATH_MAX + 160]; if (!text || snprintf(tmp, sizeof(tmp), "%s.tmp", path) >= (int)sizeof(tmp)) return -1; FILE *f = fopen(tmp, "wb"); if (!f) return -1; int ok = fputs(text, f) >= 0; if (fclose(f) != 0) ok = 0; return ok ? os_rename_replace(tmp, path) : -1; } static int find_role(const cst_agent_role_t *v, int n, const char *name) { for (int i = 0; i < n; i++) if (strcmp(v[i].name, name) == 0) return i; return -1; } typedef struct { pthread_mutex_t mu; int tasks, requests; long prompt, cached, output, cache_write; int cache_known; double cost; int priced; char currency[8]; } worker_budget_t; typedef struct { worker_budget_t *shared; int used, limit, total; } worker_gate_t; void *cst_worker_budget_new(void) { worker_budget_t *b = calloc(1, sizeof(*b)); if (b) pthread_mutex_init(&b->mu, NULL); return b; } void cst_worker_budget_free(void *p) { worker_budget_t *b = p; if (b) { pthread_mutex_destroy(&b->mu); free(b); } } void cst_worker_budget_reset(void *p) { worker_budget_t *b = p; if (!b) return; pthread_mutex_lock(&b->mu); b->tasks = b->requests = 0; b->prompt = b->cached = b->output = 0; b->cache_write = 0; b->cost = 0; b->priced = 1; b->currency[0] = 0; b->cache_known = 1; pthread_mutex_unlock(&b->mu); } char *cst_workers_budget_json(const cst_agent_config_t *cfg) { worker_budget_t *b = cfg->worker_budget; cJSON *j = cJSON_CreateObject(); if (b) { pthread_mutex_lock(&b->mu); cJSON_AddNumberToObject(j, "delegations", b->tasks); cJSON_AddNumberToObject(j, "requests", b->requests); cJSON_AddNumberToObject(j, "max_tasks", cfg->subagent_tasks); cJSON_AddNumberToObject(j, "max_requests", cfg->subagent_total); cJSON_AddNumberToObject(j, "prompt_tokens", b->prompt); cJSON_AddNumberToObject(j, "output_tokens", b->output); cJSON_AddNumberToObject(j, "cache_write_tokens", b->cache_write); if (b->priced && b->currency[0]) { cJSON_AddNumberToObject(j, "cost", b->cost); cJSON_AddStringToObject(j, "currency", b->currency); } if (b->cache_known) cJSON_AddNumberToObject(j, "cached_tokens", b->cached); else cJSON_AddNullToObject(j, "cached_tokens"); pthread_mutex_unlock(&b->mu); } char *s = cJSON_PrintUnformatted(j); cJSON_Delete(j); return s; } static int worker_request(void *p) { worker_gate_t *g = p; if (!g || !g->shared) return 0; pthread_mutex_lock(&g->shared->mu); int ok = g->used < g->limit && g->shared->requests < g->total; if (ok) { g->used++; g->shared->requests++; } pthread_mutex_unlock(&g->shared->mu); return ok; } static void workers_path(const char *ws, char *path, size_t n) { snprintf(path, n, "%s/.agent/workers.json", ws); } char *cst_workers_config_get(const char *ws) { char path[OS_PATH_MAX]; workers_path(ws, path, sizeof(path)); char *s = slurp(path, ROLE_FILE_MAX); if (!s) { char config[OS_PATH_MAX]; if (os_config_dir(config, sizeof(config)) == 0 && os_path_join(path, sizeof(path), config, "workers.json") == 0) s = slurp(path, ROLE_FILE_MAX); } return s ? s : strdup("{\"orchestrate\":false,\"max_steps\":20,\"max_tasks\":8,\"max_requests\":100,\"profiles\":{},\"roles\":{}}"); } static cJSON *workers_json(const char *ws) { char *s = cst_workers_config_get(ws); cJSON *j = s ? cJSON_Parse(s) : NULL; free(s); return j; } static const char *json_string(const cJSON *o, const char *key) { const cJSON *v = cJSON_GetObjectItemCaseSensitive(o, key); return cJSON_IsString(v) ? v->valuestring : ""; } static int json_limit(const cJSON *o, const char *key, int fallback) { const cJSON *v = cJSON_GetObjectItemCaseSensitive(o, key); return cJSON_IsNumber(v) && v->valueint > 0 ? v->valueint : fallback; } int cst_workers_config_set(const char *ws, const char *json, char *err, size_t err_sz) { cJSON *j = cJSON_Parse(json); const char *problem = NULL; if (!cJSON_IsObject(j)) problem = "expected JSON object"; const char *limits[] = {"max_steps", "max_tasks", "max_requests"}; for (int i = 0; !problem && i < 3; i++) { const cJSON *v = cJSON_GetObjectItemCaseSensitive(j, limits[i]); if (v && (!cJSON_IsNumber(v) || v->valuedouble != v->valueint || v->valueint < 1 || v->valueint > 10000)) problem = "budgets must be integers from 1 to 10000"; } const cJSON *profiles = cJSON_GetObjectItemCaseSensitive(j, "profiles"); const cJSON *roles = cJSON_GetObjectItemCaseSensitive(j, "roles"); if (!problem && ((profiles && !cJSON_IsObject(profiles)) || (roles && !cJSON_IsObject(roles)))) problem = "profiles and roles must be objects"; const cJSON *mode = cJSON_GetObjectItemCaseSensitive(j, "orchestrate"); if (mode && !cJSON_IsBool(mode)) problem = "orchestrate must be boolean"; const cJSON *p; cJSON_ArrayForEach(p, profiles) { if (!cJSON_IsObject(p) || !cst_provider_find(json_string(p, "provider")) || !json_string(p, "model")[0] || cJSON_GetObjectItemCaseSensitive(p, "api_key")) problem = "profile needs a known provider and model; use provider credentials"; } cJSON_ArrayForEach(p, roles) { if (!cJSON_IsString(p) || !cJSON_GetObjectItemCaseSensitive(profiles, p->valuestring)) problem = "role binding must name a configured profile"; } if (problem) { snprintf(err, err_sz, "%s", problem); cJSON_Delete(j); return -1; } char dir[OS_PATH_MAX], path[OS_PATH_MAX], tmp[OS_PATH_MAX + 8]; snprintf(dir, sizeof(dir), "%s/.agent", ws); workers_path(ws, path, sizeof(path)); snprintf(tmp, sizeof(tmp), "%s.tmp", path); char *s = cJSON_Print(j); cJSON_Delete(j); FILE *f = os_mkdir_p(dir, 0700) == 0 && s ? fopen(tmp, "wb") : NULL; int ok = 0; if (f) { ok = fputs(s, f) >= 0; if (fclose(f) != 0) ok = 0; } free(s); if (!ok || os_rename_replace(tmp, path) != 0) { snprintf(err, err_sz, "cannot save worker configuration"); return -1; } return 0; } void cst_workers_config_apply(cst_agent_config_t *cfg) { cJSON *j = workers_json(cfg->workspace_root); if (j) { cfg->orchestrate = cJSON_IsTrue(cJSON_GetObjectItemCaseSensitive(j, "orchestrate")); cfg->subagent_steps = json_limit(j, "max_steps", 20); cfg->subagent_tasks = json_limit(j, "max_tasks", 8); cfg->subagent_total = json_limit(j, "max_requests", 100); } cJSON_Delete(j); if (cfg->orchestrate_override) cfg->orchestrate = cfg->orchestrate_override > 0; } /* "a, b ,c" -> ",a,b,c," */ static void tool_list(const char *in, char *out, size_t sz) { size_t o = 0; out[0] = 0; if (!in || !in[0]) return; if (o + 1 < sz) out[o++] = ','; for (const char *p = in; *p && o + 2 < sz; p++) { if (*p == ' ' || *p == '\t' || *p == '[' || *p == ']' || *p == '"') continue; if (*p == ',' && out[o - 1] == ',') continue; out[o++] = *p; } if (out[o - 1] != ',' && o + 1 < sz) out[o++] = ','; out[o] = 0; if (strcmp(out, ",") == 0) out[0] = 0; } static int truthy(const char *v) { return strcmp(v, "true") == 0 || strcmp(v, "yes") == 0 || strcmp(v, "1") == 0; } static void load_dir(const char *dir, const char *source, cst_agent_role_t *v, int *n, int max) { os_dir_t *d = os_dir_open(dir); if (!d) return; const os_dirent_t *de; while ((de = os_dir_next(d)) != NULL && *n < max) { size_t len = strlen(de->name); if (de->name[0] == '.' || len < 4 || strcmp(de->name + len - 3, ".md") != 0) continue; char p[OS_PATH_MAX]; if (snprintf(p, sizeof(p), "%s/%s", dir, de->name) >= (int)sizeof(p)) continue; char *text = slurp(p, ROLE_FILE_MAX); if (!text) continue; char name[64], val[1024]; if (!cst_fm_get(text, "name", name, sizeof(name)) || !name[0]) snprintf(name, sizeof(name), "%.*s", (int)(len - 3), de->name); char clean[64]; cst_tool_sanitize_name(name, clean, sizeof(clean)); int at = find_role(v, *n, clean); if (at >= 0 && strcmp(v[at].source, "builtin") != 0) { free(text); /* the workspace one came first */ continue; } cst_agent_role_t *r = at >= 0 ? &v[at] : &v[(*n)++]; free(r->prompt); memset(r, 0, sizeof(*r)); snprintf(r->name, sizeof(r->name), "%s", clean); snprintf(r->source, sizeof(r->source), "%s", source); if (cst_fm_get(text, "description", val, sizeof(val))) snprintf(r->desc, sizeof(r->desc), "%.255s", val); if (cst_fm_get(text, "model", val, sizeof(val))) snprintf(r->model, sizeof(r->model), "%.127s", val); if (cst_fm_get(text, "profile", val, sizeof(val))) snprintf(r->profile, sizeof(r->profile), "%.63s", val); if (cst_fm_get(text, "tools", val, sizeof(val))) tool_list(val, r->tools, sizeof(r->tools)); if (cst_fm_get(text, "max_steps", val, sizeof(val))) r->max_steps = atoi(val) > 0 ? atoi(val) : 0; if (cst_fm_get(text, "readonly", val, sizeof(val)) || cst_fm_get(text, "read_only", val, sizeof(val))) r->read_only = truthy(val); const char *body = cst_fm_body(text); while (*body == '\n' || *body == '\r') body++; r->prompt = strdup(body); free(text); } os_dir_close(d); } int cst_roles_list(const char *ws, cst_agent_role_t *out, int max) { int n = 0; for (int i = 0; i < NBUILTIN && n < max; i++, n++) { memset(&out[n], 0, sizeof(out[n])); snprintf(out[n].name, sizeof(out[n].name), "%s", BUILTIN[i].name); snprintf(out[n].desc, sizeof(out[n].desc), "%s", BUILTIN[i].desc); snprintf(out[n].source, sizeof(out[n].source), "builtin"); out[n].read_only = BUILTIN[i].read_only; out[n].prompt = strdup(BUILTIN[i].prompt); } char p[OS_PATH_MAX]; if (ws && ws[0]) { snprintf(p, sizeof(p), "%.4000s/.agent/agents", ws); load_dir(p, "workspace", out, &n, max); } char cfg[OS_PATH_MAX]; if (os_config_dir(cfg, sizeof(cfg)) == 0) { snprintf(p, sizeof(p), "%.4000s/agents", cfg); load_dir(p, "user", out, &n, max); } /* custom roles after the built-ins, by name */ cJSON *settings = workers_json(ws ? ws : "."); const cJSON *bindings = cJSON_GetObjectItemCaseSensitive(settings, "roles"); for (int i = 0; i < n; i++) { const char *profile = json_string(bindings, out[i].name); if (*profile) snprintf(out[i].profile, sizeof(out[i].profile), "%s", profile); } cJSON_Delete(settings); for (int i = NBUILTIN + 1; i < n; i++) for (int j = i; j > NBUILTIN && strcmp(out[j - 1].name, out[j].name) > 0; j--) { cst_agent_role_t t = out[j]; out[j] = out[j - 1]; out[j - 1] = t; } return n; } void cst_roles_free(cst_agent_role_t *v, int n) { for (int i = 0; i < n; i++) { free(v[i].prompt); v[i].prompt = NULL; } } char *cst_subagent_index(const char *ws) { cst_agent_role_t *v = calloc(CST_ROLES_MAX, sizeof(*v)); if (!v) return NULL; int n = cst_roles_list(ws, v, CST_ROLES_MAX); cst_buf_t b = {0}; cst_buf_puts(&b, "\nSub-agents (task tool, agent=; default " "general):\n"); for (int i = 0; i < n && b.len < INDEX_MAX; i++) cst_buf_printf(&b, "- %s%s: %.200s [model profile: %s]\n", v[i].name, v[i].read_only ? " (read-only)" : "", v[i].desc, v[i].profile[0] ? v[i].profile : "inherits parent"); cst_roles_free(v, n); free(v); return cst_buf_steal(&b); } /* ---------- transcripts ---------- */ static int sub_dir(const char *ws, const char *parent_id, char *out, size_t sz) { char dir[OS_PATH_MAX]; if (!parent_id || !parent_id[0] || strchr(parent_id, '/') || strchr(parent_id, '\\') || cst_agent_store_dir(ws, dir, sizeof(dir)) != 0) return -1; return snprintf(out, sz, "%s/%s", dir, parent_id) < (int)sz ? 0 : -1; } int cst_subagent_path(const char *ws, const char *parent_id, const char *file, char *out, size_t sz) { char dir[OS_PATH_MAX]; if (!file || strchr(file, '/') || strchr(file, '\\') || sub_dir(ws, parent_id, dir, sizeof(dir)) != 0) return -1; return snprintf(out, sz, "%s/%s.jsonl", dir, file) < (int)sz ? 0 : -1; } /* Reserve the next sub-.jsonl. */ static int next_file(const char *dir, char *file, size_t fsz) { pthread_mutex_lock(&g_seq_mu); int best = 0; os_dir_t *d = os_dir_open(dir); if (d) { const os_dirent_t *de; while ((de = os_dir_next(d)) != NULL) { int k = 0; if (sscanf(de->name, "sub-%d.jsonl", &k) == 1 && k > best) best = k; } os_dir_close(d); } int rc = -1; for (int k = best + 1; k < best + 100; k++) { char p[OS_PATH_MAX]; snprintf(file, fsz, "sub-%d", k); if (snprintf(p, sizeof(p), "%s/%s.jsonl", dir, file) >= (int)sizeof(p)) break; struct stat st; if (stat(p, &st) == 0) continue; FILE *fp = fopen(p, "ab"); if (fp) { fclose(fp); rc = 0; } break; } pthread_mutex_unlock(&g_seq_mu); return rc; } static void write_meta(const char *dir, const char *file, const char *role, const char *desc, const char *status, int steps, int tools, double ms) { char p[OS_PATH_MAX]; if (snprintf(p, sizeof(p), "%s/%s.json", dir, file) >= (int)sizeof(p)) return; cJSON *o = cJSON_CreateObject(); cJSON_AddStringToObject(o, "role", role); cJSON_AddStringToObject(o, "description", desc); cJSON_AddStringToObject(o, "status", status); cJSON_AddNumberToObject(o, "steps", steps); cJSON_AddNumberToObject(o, "tools", tools); cJSON_AddNumberToObject(o, "ms", ms); char *txt = cJSON_PrintUnformatted(o); cJSON_Delete(o); save_atomic(p, txt); free(txt); } int cst_subagent_list(const char *ws, const char *parent_id, cst_subagent_rec_t *out, int max) { char dir[OS_PATH_MAX]; if (sub_dir(ws, parent_id, dir, sizeof(dir)) != 0) return 0; int n = 0; os_dir_t *d = os_dir_open(dir); if (!d) return 0; const os_dirent_t *de; while ((de = os_dir_next(d)) != NULL && n < max) { int k = 0; char tail[8] = ""; if (sscanf(de->name, "sub-%d.%7s", &k, tail) != 2 || strcmp(tail, "json") != 0) continue; char p[OS_PATH_MAX]; if (snprintf(p, sizeof(p), "%s/%s", dir, de->name) >= (int)sizeof(p)) continue; char *txt = slurp(p, 64 * 1024); cJSON *o = txt ? cJSON_Parse(txt) : NULL; free(txt); cst_subagent_rec_t *r = &out[n]; memset(r, 0, sizeof(*r)); snprintf(r->file, sizeof(r->file), "sub-%d", k); const cJSON *it; if ((it = cJSON_GetObjectItemCaseSensitive(o, "role")) && it->valuestring) snprintf(r->role, sizeof(r->role), "%s", it->valuestring); if ((it = cJSON_GetObjectItemCaseSensitive(o, "description")) && it->valuestring) snprintf(r->desc, sizeof(r->desc), "%s", it->valuestring); if ((it = cJSON_GetObjectItemCaseSensitive(o, "status")) && it->valuestring) snprintf(r->status, sizeof(r->status), "%s", it->valuestring); snprintf(r->provider, sizeof(r->provider), "%s", json_string(o, "provider")); snprintf(r->model, sizeof(r->model), "%s", json_string(o, "model")); #define READ_NUMBER(field) do { const cJSON *v = cJSON_GetObjectItemCaseSensitive(o, #field); if (cJSON_IsNumber(v)) r->field = v->valuedouble; } while (0) READ_NUMBER(requests); READ_NUMBER(max_steps); READ_NUMBER(prompt_tokens); READ_NUMBER(cached_tokens); READ_NUMBER(cache_write_tokens); READ_NUMBER(output_tokens); READ_NUMBER(cost); #undef READ_NUMBER r->priced = cJSON_IsNumber(cJSON_GetObjectItemCaseSensitive(o, "cost")); r->cache_known = cJSON_IsNumber(cJSON_GetObjectItemCaseSensitive(o, "cached_tokens")); snprintf(r->currency, sizeof(r->currency), "%s", json_string(o, "currency")); if (strcmp(r->status, "running") == 0) { char transcript[OS_PATH_MAX]; if (cst_subagent_path(ws, parent_id, r->file, transcript, sizeof(transcript)) == 0) { if (!worker_active(transcript)) snprintf(r->status, sizeof(r->status), "interrupted"); else worker_live(transcript, r); } } if ((it = cJSON_GetObjectItemCaseSensitive(o, "steps"))) r->steps = it->valueint; if ((it = cJSON_GetObjectItemCaseSensitive(o, "tools"))) r->tools = it->valueint; if ((it = cJSON_GetObjectItemCaseSensitive(o, "ms"))) r->ms = it->valuedouble; cJSON_Delete(o); struct stat st; if (stat(p, &st) == 0) r->mtime = (long long)st.st_mtime; n++; } os_dir_close(d); /* newest first: higher number */ for (int i = 1; i < n; i++) for (int j = i; j > 0 && atoi(out[j - 1].file + 4) < atoi(out[j].file + 4); j--) { cst_subagent_rec_t t = out[j]; out[j] = out[j - 1]; out[j - 1] = t; } return n; } /* ---------- the task tool ---------- */ typedef struct { const cst_agent_hooks_t *ph; const ctl_tool_call_t *pcall; const char *role; const char *ws; char last[2][80]; int ntools; volatile int *parent_cancel; cst_agent_t *child; atomic_int done; } sub_run_t; static const char *arg(const ctl_tool_call_t *c, const char *key) { for (int i = 0; i < c->nargs; i++) if (strcmp(c->args[i].key, key) == 0) return c->args[i].value ? c->args[i].value : ""; return NULL; } /* "grep foo" / "read_file src/x.c" for the progress line. */ static void step_summary(const ctl_tool_call_t *c, const char *ws, char *out, size_t sz) { static const char *keys[] = {"path", "pattern", "cmd", "command", "url", "query", "name", "file"}; const char *v = NULL; for (size_t i = 0; i < sizeof(keys) / sizeof(keys[0]) && !v; i++) v = arg(c, keys[i]); char one[48] = ""; size_t wl = ws ? strlen(ws) : 0; if (v && wl && strncmp(v, ws, wl) == 0 && v[wl] == '/' && v[wl + 1]) v += wl + 1; if (v) { size_t k = 0; for (; v[k] && v[k] != '\n' && k < 40; k++) one[k] = v[k]; while (k > 0 && ((unsigned char)one[k] & 0xC0) == 0x80 && v[k]) k--; one[k] = 0; } snprintf(out, sz, "%.30s%s%.40s", c->name, one[0] ? " " : "", one); } static void sub_tool_event(const char *phase, const ctl_tool_call_t *call, const char *detail, void *ud) { (void)detail; sub_run_t *r = ud; if (!call || strcmp(phase, "run") != 0) return; r->ntools++; memcpy(r->last[0], r->last[1], sizeof(r->last[0])); step_summary(call, r->ws, r->last[1], sizeof(r->last[1])); if (!r->ph || !r->ph->on_tool_event) return; char line[256]; snprintf(line, sizeof(line), "%s%s%s · %s", r->role, r->last[0][0] ? " · " : "", r->last[0], r->last[1]); pthread_mutex_lock(&g_hook_mu); r->ph->on_tool_event("progress", r->pcall, line, r->ph->userdata); pthread_mutex_unlock(&g_hook_mu); } static int sub_permission(const char *tool, const char *detail, const char *preview, void *ud) { sub_run_t *r = ud; if (!r->ph || !r->ph->on_permission) return CST_PERM_DENY; cst_buf_t d = {0}; size_t sbl = strlen(CST_PERM_SANDBOX); if (detail && strncmp(detail, CST_PERM_SANDBOX, sbl) == 0) { cst_buf_printf(&d, "%s[%s] %s", CST_PERM_SANDBOX, r->role, detail + sbl); } else { cst_buf_printf(&d, "[%s] %s", r->role, detail ? detail : tool); } /* one question at a time, whichever sub-agent asks */ pthread_mutex_lock(&g_hook_mu); int ans = r->ph->on_permission(tool, d.buf ? d.buf : tool, preview, r->ph->userdata); pthread_mutex_unlock(&g_hook_mu); cst_buf_free(&d); return ans; } static void *watch_main(void *p) { sub_run_t *r = p; while (!atomic_load(&r->done)) { if (r->parent_cancel && *r->parent_cancel) cst_agent_request_cancel(r->child); struct timespec ts = {0, 50 * 1000000L}; nanosleep(&ts, NULL); } return NULL; } static int out_text(cst_tool_result_t *r, int code, const char *text) { r->exit_code = code; r->out = strdup(text); r->out_len = r->out ? strlen(r->out) : 0; return 0; } static unsigned long worker_hash(const char *s) { unsigned long h = 5381; for (; *s; s++) h = h * 33 + (unsigned char)*s; return h; } static pthread_mutex_t worker_mu = PTHREAD_MUTEX_INITIALIZER; static struct { char path[OS_PATH_MAX]; cst_agent_t *child; worker_gate_t *gate; } active_workers[64]; static void worker_live(const char *path, cst_subagent_rec_t *r) { pthread_mutex_lock(&worker_mu); for (int i = 0; i < 64; i++) { if (active_workers[i].gate && !strcmp(active_workers[i].path, path)) { worker_gate_t *g = active_workers[i].gate; pthread_mutex_lock(&g->shared->mu); r->requests = g->used; r->max_steps = g->limit; pthread_mutex_unlock(&g->shared->mu); } } pthread_mutex_unlock(&worker_mu); } static int worker_active(const char *path) { int found = 0; pthread_mutex_lock(&worker_mu); for (int i = 0; i < 64; i++) if (active_workers[i].child && strcmp(active_workers[i].path, path) == 0) found = 1; pthread_mutex_unlock(&worker_mu); return found; } int cst_worker_cancel(const char *ws, const char *pid, const char *id) { char path[OS_PATH_MAX]; if (cst_subagent_path(ws, pid, id, path, sizeof(path)) != 0) return -1; int found = 0; pthread_mutex_lock(&worker_mu); for (int i = 0; i < 64; i++) { if (active_workers[i].child && strcmp(active_workers[i].path, path) == 0) { cst_agent_request_cancel(active_workers[i].child); found = 1; } } pthread_mutex_unlock(&worker_mu); return found ? 0 : -1; } static cJSON *worker_snapshot(const char *path, cst_agent_config_t *cc, const char *role, int save) { char p[OS_PATH_MAX + 16]; snprintf(p, sizeof(p), "%s.worker", path); cJSON *j; if (!save) { char *s = slurp(p, ROLE_FILE_MAX); j = s ? cJSON_Parse(s) : NULL; free(s); return j; } j = cJSON_CreateObject(); cJSON_AddStringToObject(j, "role", role); cJSON_AddStringToObject(j, "workspace", cc->workspace_root); cJSON_AddStringToObject(j, "scope", cc->worker_scope); cJSON_AddStringToObject(j, "provider", cc->provider); cJSON_AddStringToObject(j, "api_base", cc->api_base); cJSON_AddStringToObject(j, "model", cc->model); cJSON_AddStringToObject(j, "effort", cc->effort); cJSON_AddStringToObject(j, "tools", cc->tools_only); cJSON_AddStringToObject(j, "tools_off", cc->tools_off); cJSON_AddNumberToObject(j, "protocol", cc->protocol); cJSON_AddNumberToObject(j, "tool_profile", cc->tool_profile); cJSON_AddStringToObject(j, "prompt", cc->role_prompt ? cc->role_prompt : ""); cJSON_AddBoolToObject(j, "readonly", cc->read_only); cJSON_AddNumberToObject(j, "temperature", cc->temperature); cJSON_AddNumberToObject(j, "top_p", cc->top_p); cJSON_AddNumberToObject(j, "max_tokens", cc->max_tokens); char *s = cJSON_PrintUnformatted(j); int ok = save_atomic(p, s) == 0; free(s); if (!ok) { cJSON_Delete(j); return NULL; } return j; } static void worker_usage(const char *dir, const char *file, const cst_agent_config_t *cc, cst_agent_t *child, int requests) { char path[OS_PATH_MAX + 128]; snprintf(path, sizeof(path), "%s/%s.json", dir, file); char *s = slurp(path, ROLE_FILE_MAX); cJSON *j = s ? cJSON_Parse(s) : NULL; free(s); if (!j) return; cJSON_AddStringToObject(j, "provider", cc->provider); cJSON_AddStringToObject(j, "model", cc->model); cJSON_AddNumberToObject(j, "requests", requests); cJSON_AddNumberToObject(j, "max_steps", cc->max_steps); cst_usage_t u; memset(&u, 0, sizeof(u)); if (child) cst_agent_usage(child, NULL, &u); cJSON_AddNumberToObject(j, "prompt_tokens", u.prompt); if (u.requests > 0 && u.cache_known_requests == u.requests) { cJSON_AddNumberToObject(j, "cached_tokens", u.cached); cJSON_AddNumberToObject(j, "cache_write_tokens", u.write); } cJSON_AddNumberToObject(j, "output_tokens", u.out); if (u.priced) { cJSON_AddNumberToObject(j, "cost", u.cost); cJSON_AddStringToObject(j, "currency", u.currency); } char *text = cJSON_PrintUnformatted(j); cJSON_Delete(j); save_atomic(path, text); snprintf(path, sizeof(path), "%s/%s.runs.jsonl", dir, file); FILE *f = text ? fopen(path, "ab") : NULL; if (f) { fputs(text, f); fputc('\n', f); fclose(f); } free(text); } static int task_run(const cst_tool_ctx_t *ctx, const ctl_tool_call_t *c, cst_tool_result_t *r, void *ud) { (void)ud; if (!ctx->agent) return out_text(r, 1, "error: task needs a running agent"); if (cst_parliament_is_active()) return out_text(r, 1, "error: parliament mode replaces in-process workers; " "publish parliament_note and coordinate via peer seats"); const cst_agent_config_t *pcfg = cst_agent_config(ctx->agent); if (pcfg->depth > 0) return out_text(r, 1, "error: a sub-agent cannot start sub-agents; do " "the work yourself"); const char *desc = arg(c, "description"); const char *prompt = arg(c, "prompt"); const char *want = arg(c, "agent"); const char *worker_id = arg(c, "worker_id"); if (!want || !want[0]) want = "general"; if (!prompt || !prompt[0]) return out_text(r, 1, "error: prompt is required"); if (!desc || !desc[0]) desc = want; cst_agent_role_t *roles = calloc(CST_ROLES_MAX, sizeof(*roles)); if (!roles) return out_text(r, 1, "error: out of memory"); char pid[256], dir[OS_PATH_MAX], file[64], path[OS_PATH_MAX]; cst_agent_session_id(ctx->agent, pid, sizeof(pid)); cJSON *snapshot = NULL; if (worker_id && worker_id[0]) { int seq = 0, end = 0; if (sscanf(worker_id, "sub-%d%n", &seq, &end) != 1 || seq < 1 || worker_id[end] || cst_subagent_path(ctx->workspace_root, pid, worker_id, path, sizeof(path)) != 0) { free(roles); return out_text(r, 1, "error: invalid worker_id"); } snapshot = worker_snapshot(path, NULL, NULL, 0); if (!snapshot) { free(roles); return out_text(r, 1, "error: worker not found"); } want = json_string(snapshot, "role"); } int nr = cst_roles_list(ctx->workspace_root, roles, CST_ROLES_MAX); int ri = find_role(roles, nr, want); if (ri < 0 && snapshot && nr < CST_ROLES_MAX) { ri = nr++; memset(&roles[ri], 0, sizeof(roles[ri])); snprintf(roles[ri].name, sizeof(roles[ri].name), "%s", want); roles[ri].prompt = strdup(json_string(snapshot, "prompt")); } if (ri < 0) { cst_buf_t b = {0}; cst_buf_printf(&b, "error: no agent named \"%s\" (available:", want); for (int i = 0; i < nr; i++) cst_buf_printf(&b, "%s %s", i ? "," : "", roles[i].name); cst_buf_puts(&b, ")"); cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); r->exit_code = 1; r->out_len = b.len; r->out = cst_buf_steal(&b); return 0; } const cst_agent_role_t *role = &roles[ri]; if (sub_dir(ctx->workspace_root, pid, dir, sizeof(dir)) != 0 || os_mkdir_p(dir, 0755) != 0 || (!snapshot && next_file(dir, file, sizeof(file)) != 0) || (snapshot && snprintf(file, sizeof(file), "%s", worker_id) < 0) || snprintf(path, sizeof(path), "%s/%s.jsonl", dir, file) >= (int)sizeof(path)) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: cannot create the sub-agent transcript"); } cst_agent_config_t cc = *pcfg; cc.depth = pcfg->depth + 1; cc.task_revision = cst_runtime_revision(pcfg->runtime); const char *isolated = arg(c, "_worktree"); if (isolated && isolated[0]) snprintf(cc.workspace_root, sizeof(cc.workspace_root), "%s", isolated); const char *scope = arg(c, "_scope"); if (scope) snprintf(cc.worker_scope, sizeof(cc.worker_scope), "%s", scope); cc.plan_mode = 0; cc.read_only = role->read_only || pcfg->plan_mode || pcfg->read_only; snprintf(cc.tools_only, sizeof(cc.tools_only), "%s", role->tools); cst_tool_set_disabled(&cc, "ask_user", 1); cc.role_prompt = role->prompt; if (role->model[0]) snprintf(cc.model, sizeof(cc.model), "%s", role->model); cc.max_steps = pcfg->subagent_steps > 0 ? pcfg->subagent_steps : 20; if (role->max_steps > 0 && role->max_steps < cc.max_steps) cc.max_steps = role->max_steps; const char *requested_steps = arg(c, "max_steps"); if (requested_steps) { char *end; long v = strtol(requested_steps, &end, 10); if (*end || v < 1 || v > cc.max_steps) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: max_steps must be within the configured limit"); } cc.max_steps = (int)v; } cJSON *settings = workers_json(ctx->workspace_root); const cJSON *profiles = cJSON_GetObjectItemCaseSensitive(settings, "profiles"); const cJSON *profile = cJSON_GetObjectItemCaseSensitive(profiles, role->profile); if (!snapshot && role->profile[0]) { const cst_provider_t *provider = cst_provider_find(json_string(profile, "provider")); if (!provider || !json_string(profile, "model")[0]) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(settings); return out_text(r, 1, "error: role model profile is missing or invalid"); } snprintf(cc.provider, sizeof(cc.provider), "%s", provider->name); snprintf(cc.api_base, sizeof(cc.api_base), "%s", provider->api_base); snprintf(cc.model, sizeof(cc.model), "%s", json_string(profile, "model")); snprintf(cc.effort, sizeof(cc.effort), "%s", json_string(profile, "effort")); cc.temperature = 0; cc.top_p = 0; cc.max_tokens = 0; const cJSON *v = cJSON_GetObjectItemCaseSensitive(profile, "temperature"); if (cJSON_IsNumber(v)) cc.temperature = v->valuedouble; v = cJSON_GetObjectItemCaseSensitive(profile, "top_p"); if (cJSON_IsNumber(v)) cc.top_p = v->valuedouble; cc.max_tokens = json_limit(profile, "max_tokens", 0); cc.api_key[0] = 0; if (!strcmp(cc.provider, pcfg->provider) && !strcmp(cc.api_base, pcfg->api_base)) snprintf(cc.api_key, sizeof(cc.api_key), "%s", pcfg->api_key); if (!cc.api_key[0] && cst_creds_resolve(cc.provider, cc.api_key, sizeof(cc.api_key), NULL) != 0) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(settings); return out_text(r, 1, "error: no provider credentials for role profile"); } } cJSON_Delete(settings); if (snapshot) { const char *saved_workspace = json_string(snapshot, "workspace"); if (*saved_workspace) snprintf(cc.workspace_root, sizeof(cc.workspace_root), "%s", saved_workspace); snprintf(cc.worker_scope, sizeof(cc.worker_scope), "%s", json_string(snapshot, "scope")); snprintf(cc.provider, sizeof(cc.provider), "%s", json_string(snapshot, "provider")); snprintf(cc.api_base, sizeof(cc.api_base), "%s", json_string(snapshot, "api_base")); snprintf(cc.model, sizeof(cc.model), "%s", json_string(snapshot, "model")); snprintf(cc.effort, sizeof(cc.effort), "%s", json_string(snapshot, "effort")); snprintf(cc.tools_only, sizeof(cc.tools_only), "%s", json_string(snapshot, "tools")); const char *off = json_string(snapshot, "tools_off"); if (strlen(cc.tools_off) + strlen(off) < sizeof(cc.tools_off)) strcat(cc.tools_off, off); else { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: tool restriction list exceeds capacity"); } const cJSON *proto = cJSON_GetObjectItemCaseSensitive(snapshot, "protocol"); if (cJSON_IsNumber(proto)) cc.protocol = proto->valueint; const cJSON *tp = cJSON_GetObjectItemCaseSensitive(snapshot, "tool_profile"); if (cJSON_IsNumber(tp)) cc.tool_profile = tp->valueint; cc.role_prompt = json_string(snapshot, "prompt"); cc.read_only = cc.read_only || cJSON_IsTrue(cJSON_GetObjectItemCaseSensitive(snapshot, "readonly")); if (strcmp(cc.provider, pcfg->provider) != 0 || strcmp(cc.api_base, pcfg->api_base) != 0) { cc.api_key[0] = 0; if (cst_creds_resolve(cc.provider, cc.api_key, sizeof(cc.api_key), NULL) != 0) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: no credentials for resumed worker"); } } const cJSON *v = cJSON_GetObjectItemCaseSensitive(snapshot, "temperature"); cc.temperature = cJSON_IsNumber(v) ? v->valuedouble : 0; v = cJSON_GetObjectItemCaseSensitive(snapshot, "top_p"); cc.top_p = cJSON_IsNumber(v) ? v->valuedouble : 0; cc.max_tokens = json_limit(snapshot, "max_tokens", 0); } worker_budget_t *budget = pcfg->worker_budget; if (pcfg->tools_only[0]) { cst_agent_config_t permitted = cc; cst_buf_t names = {0}; for (int i = 0; i < cst_tool_count(); i++) { const cst_tool_def_t *tool = cst_tool_at(i); if (cst_tool_active(pcfg, tool) && cst_tool_active(&permitted, tool)) cst_buf_printf(&names, ",%s,", tool->name); } snprintf(cc.tools_only, sizeof(cc.tools_only), "%s", names.buf ? names.buf : ",none,"); cst_buf_free(&names); } if (!snapshot) { cst_buf_t tools = {0}; for (int i = 0; i < cst_tool_count(); i++) { const cst_tool_def_t *tool = cst_tool_at(i); if (cst_tool_active(&cc, tool)) cst_buf_printf(&tools, ",%s,", tool->name); } if (tools.len >= sizeof(cc.tools_only)) { cst_buf_free(&tools); cst_roles_free(roles, nr); free(roles); return out_text(r, 1, "error: worker tool list exceeds capacity"); } snprintf(cc.tools_only, sizeof(cc.tools_only), "%s", tools.buf ? tools.buf : ",none,"); cst_buf_free(&tools); } if (!budget) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: no shared worker budget"); } pthread_mutex_lock(&budget->mu); int allowed = budget->tasks < (pcfg->subagent_tasks > 0 ? pcfg->subagent_tasks : 8); if (allowed) budget->tasks++; pthread_mutex_unlock(&budget->mu); if (!allowed) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: delegation budget exhausted"); } worker_gate_t gate = {budget, 0, cc.max_steps, pcfg->subagent_total > 0 ? pcfg->subagent_total : 100}; cc.request_gate = worker_request; cc.request_gate_ud = &gate; char cache_input[8192]; snprintf(cache_input, sizeof(cache_input), "%s|%s|%s|%s|%s|v4", cc.workspace_root, cc.provider, cc.model, role->name, cc.tools_only); snprintf(cc.cache_namespace, sizeof(cc.cache_namespace), "worker-%lx", worker_hash(cache_input)); if (!snapshot) { snapshot = worker_snapshot(path, &cc, role->name, 1); if (!snapshot) { cst_roles_free(roles, nr); free(roles); return out_text(r, 1, "error: cannot persist worker configuration"); } } snprintf(cc.session_path, sizeof(cc.session_path), "%s", path); cst_agent_undo_key(ctx->agent, cc.undo_sid, sizeof(cc.undo_sid), cc.undo_tag, sizeof(cc.undo_tag)); volatile int *parent_cancel = os_get_cancel_flag(); sub_run_t run; memset(&run, 0, sizeof(run)); run.ph = ctx->hooks; run.pcall = c; run.role = role->name; run.ws = ctx->workspace_root; run.parent_cancel = parent_cancel; pthread_mutex_lock(&worker_mu); int slot = -1; for (int i = 0; i < 64; i++) { if (active_workers[i].path[0] && strcmp(active_workers[i].path, path) == 0) { slot = -2; break; } if (!active_workers[i].path[0] && slot < 0) slot = i; } if (slot >= 0) { snprintf(active_workers[slot].path, sizeof(active_workers[slot].path), "%s", path); } pthread_mutex_unlock(&worker_mu); if (slot < 0) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); return out_text(r, 1, "error: worker already busy or worker capacity reached"); } run.child = cst_agent_create(&cc); pthread_mutex_lock(&worker_mu); active_workers[slot].child = run.child; active_workers[slot].gate = run.child ? &gate : NULL; if (!run.child) active_workers[slot].path[0] = 0; pthread_mutex_unlock(&worker_mu); if (!run.child) { cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); os_set_cancel_flag(parent_cancel); return out_text(r, 1, "error: cannot start the sub-agent"); } write_meta(dir, file, role->name, desc, "running", 0, 0, 0); worker_usage(dir, file, &cc, NULL, 0); cst_agent_hooks_t h; memset(&h, 0, sizeof(h)); h.on_tool_event = sub_tool_event; h.on_permission = sub_permission; h.userdata = &run; if (run.ph && run.ph->on_tool_event) { char line[160]; snprintf(line, sizeof(line), "%s · starting", role->name); pthread_mutex_lock(&g_hook_mu); run.ph->on_tool_event("progress", c, line, run.ph->userdata); pthread_mutex_unlock(&g_hook_mu); } pthread_t watcher; int watching = pthread_create(&watcher, NULL, watch_main, &run) == 0; double t0 = os_now_ms(); char err[512] = ""; int rc = cst_agent_run_turn_ex(run.child, prompt, &h, err, sizeof(err)); double ms = os_now_ms() - t0; atomic_store(&run.done, 1); if (watching) pthread_join(watcher, NULL); int cancelled = (parent_cancel && *parent_cancel) || (rc != 0 && strstr(err, "cancel") != NULL); /* the report is the last assistant text after the prompt */ cst_session_t *s = cst_agent_session(run.child); int n = cst_session_count(s), steps = 0; const char *final = NULL; int seen_user = 0; /* Only report the newest assignment when resuming a persistent worker. */ int last_user = -1; for (int i = 0; i < n; i++) { const cst_message_t *m = cst_session_get(s, i); if (m->role == CST_ROLE_USER && !m->tool_call_id[0] && m->content && strcmp(m->content, prompt) == 0) last_user = i; } for (int i = 0; i < n; i++) { if (i < last_user) continue; const cst_message_t *m = cst_session_get(s, i); if (m->role == CST_ROLE_USER && !m->tool_call_id[0] && !seen_user) { seen_user = 1; continue; } if (!seen_user || m->role != CST_ROLE_ASSISTANT) continue; steps++; if (m->content && m->content[0] && !m->tool_calls_json) final = m->content; } long tokens = cst_agent_last_prompt_tokens(run.child); cst_buf_t b = {0}; if (rc != 0) { int budget = err[0] && (strstr(err, "budget:") || strstr(err, "budget exhausted")); if (budget) { cst_buf_puts(&b, "error: worker request budget exhausted\n"); if (err[0]) cst_buf_printf(&b, "detail: %s\n", err); cst_buf_puts(&b, "hint: raise .agent/workers.json max_requests, or " "finish the remaining work yourself with local tools.\n"); } else cst_buf_printf(&b, "error: sub-agent %s: %s\n", cancelled ? "cancelled" : "failed", err[0] ? err : "unknown error"); } if (final) { while (*final == ' ' || *final == '\n') final++; cst_buf_puts(&b, final); } else if (rc == 0) { cst_buf_puts(&b, "(the sub-agent ended without a final report; its " "transcript has the details)"); } else { for (int i = n - 1; i > last_user; i--) { const cst_message_t *m = cst_session_get(s, i); if (m->role == CST_ROLE_TOOL && m->content && m->content[0]) { size_t len = strlen(m->content); if (len > 2048) len = cst_utf8_floor(m->content, 2048); cst_buf_puts(&b, "\nLast tool result (partial task, verify before continuing):\n"); cst_buf_append(&b, m->content, len); break; } } } while (b.len && (b.buf[b.len - 1] == '\n' || b.buf[b.len - 1] == ' ')) b.buf[--b.len] = 0; char tok[32] = ""; if (tokens > 0) snprintf(tok, sizeof(tok), " · %.1fk tokens", (double)tokens / 1000.0); cst_buf_printf(&b, "\n\n[task %s · %d step%s · %d tool call%s · %.1f s%s · " "transcript .agent/sessions/%s/%s.jsonl]", role->name, steps, steps == 1 ? "" : "s", run.ntools, run.ntools == 1 ? "" : "s", ms / 1000.0, tok, pid, file); write_meta(dir, file, role->name, desc, rc == 0 ? "done" : cancelled ? "cancelled" : "error", steps, run.ntools, ms); worker_usage(dir, file, &cc, run.child, gate.used); cst_usage_t usage; memset(&usage, 0, sizeof(usage)); cst_agent_usage(run.child, NULL, &usage); pthread_mutex_lock(&budget->mu); budget->prompt += usage.prompt; budget->cached += usage.cached; budget->output += usage.out; budget->cache_write += usage.write; if (usage.requests) { if (!budget->currency[0]) snprintf(budget->currency, sizeof(budget->currency), "%s", usage.currency); if (usage.priced && !strcmp(budget->currency, usage.currency)) budget->cost += usage.cost; else budget->priced = 0; } if (usage.cache_known_requests != usage.requests) budget->cache_known = 0; int remaining = gate.total - budget->requests; pthread_mutex_unlock(&budget->mu); cst_buf_printf(&b, "\n[worker_id=%s; model=%s/%s; requests=%d/%d]", file, cc.provider, cc.model, gate.used, gate.limit); cst_buf_printf(&b, "\n[shared worker requests remaining=%d]", remaining); pthread_mutex_lock(&worker_mu); active_workers[slot].child = NULL; active_workers[slot].gate = NULL; active_workers[slot].path[0] = 0; pthread_mutex_unlock(&worker_mu); cst_agent_destroy(run.child); /* run_turn and destroy re-pointed this thread's cancel flag */ os_set_cancel_flag(parent_cancel); cst_roles_free(roles, nr); free(roles); cJSON_Delete(snapshot); r->exit_code = rc == 0 ? 0 : cancelled ? 130 : 1; r->out_len = b.len; r->out = cst_buf_steal(&b); return 0; } /* Read-only roles may run side by side. */ typedef struct { cst_tool_ctx_t ctx; cst_agent_hooks_t hooks; ctl_tool_call_t call; char worktree[OS_PATH_MAX]; } worker_job_t; static void worker_job_free(void *p) { worker_job_t *j = p; cst_tool_call_clear(&j->call); free(j); } static int worker_job_run(void *p, volatile int *cancel, cst_tool_result_t *out) { worker_job_t *j = p; os_set_cancel_flag(cancel); int rc = task_run(&j->ctx, &j->call, out, NULL); if (j->worktree[0]) { cst_buf_t b = {0}; cst_buf_printf(&b, "%s\n[isolated worktree=%s; use worker_diff then worker_integrate to accept changes]", out->out ? out->out : "", j->worktree); free(out->out); out->out_len = b.len; out->out = cst_buf_steal(&b); } return rc; } static int async_worker_tool(const cst_tool_ctx_t *ctx, const ctl_tool_call_t *c, cst_tool_result_t *out, void *ud) { (void)ud; if (!ctx->agent) return out_text(out, 1, "error: no agent"); const cst_agent_config_t *cfg = cst_agent_config(ctx->agent); if (cfg->depth || !cfg->runtime) return out_text(out, 1, "error: only the main agent can schedule workers"); if (cst_parliament_is_active()) return out_text(out, 1, "error: parliament mode replaces in-process workers; " "use parliament_note and Realm peer seats"); if (strcmp(c->name, "worker_start")) { const char *wt = arg(c, "worktree"); int code; if (wt && cst_runtime_resource_busy(cfg->runtime, wt)) return out_text(out, 1, "error: worker is still using this worktree; wait before review/integration"); char *text = wt ? cst_worktree_diff(ctx->workspace_root, wt, !strcmp(c->name, "worker_integrate"), &code) : NULL; if (!text) return out_text(out, 1, "error: worktree is required"); out->out = text; out->out_len = strlen(text); out->exit_code = code; return 0; } if (!cst_runtime_check(cfg->runtime, cst_runtime_revision(cfg->runtime), 0, 1)) return out_text(out, 1, "error: delegation is disabled"); const char *kind = arg(c, "kind"); if (!kind || (strcmp(kind, "read") && strcmp(kind, "edit") && strcmp(kind, "external"))) return out_text(out, 1, "error: kind must be read, edit, or external"); if (!cst_runtime_check(cfg->runtime, cst_runtime_revision(cfg->runtime), strcmp(kind, "read") != 0, 1)) return out_text(out, 1, "error: task controls prohibit this worker kind"); worker_job_t *j = calloc(1, sizeof(*j)); if (!j) return -1; j->ctx = *ctx; j->ctx.set_todo = NULL; j->ctx.todo_ud = NULL; if (ctx->hooks) { j->hooks = *ctx->hooks; j->ctx.hooks = &j->hooks; } cJSON *a = cJSON_CreateObject(); const char *prompt = arg(c, "prompt"); const char *description = arg(c, "description"); if (!prompt || !*prompt) { cJSON_Delete(a); free(j); return out_text(out, 1, "error: prompt required"); } cJSON_AddStringToObject(a, "prompt", prompt); cJSON_AddStringToObject(a, "description", description ? description : kind); /* Typed routing prevents read-only investigations silently using Pro. */ cJSON_AddStringToObject(a, "agent", !strcmp(kind, "read") ? "explore" : "general"); const char *resume = arg(c, "worker_id"); if (resume && *resume) cJSON_AddStringToObject(a, "worker_id", resume); if (resume && *resume) { char sid[256], transcript[OS_PATH_MAX]; cst_agent_session_id(ctx->agent, sid, sizeof(sid)); cJSON *saved = cst_subagent_path(ctx->workspace_root, sid, resume, transcript, sizeof(transcript)) == 0 ? worker_snapshot(transcript, NULL, NULL, 0) : NULL; if (!saved || (!strcmp(kind, "read") && !cJSON_IsTrue(cJSON_GetObjectItemCaseSensitive(saved, "readonly")))) { cJSON_Delete(saved); cJSON_Delete(a); free(j); return out_text(out, 1, "error: invalid worker resume or mismatched task kind"); } if (!strcmp(kind, "edit")) { snprintf(j->worktree, sizeof(j->worktree), "%s", json_string(saved, "workspace")); char prefix[OS_PATH_MAX]; snprintf(prefix, sizeof(prefix), "%s/.agent/v2/wt-", ctx->workspace_root); if (strncmp(j->worktree, prefix, strlen(prefix)) || !*json_string(saved, "scope")) { cJSON_Delete(saved); cJSON_Delete(a); free(j); return out_text(out, 1, "error: edit resume requires an isolated scoped worker"); } } cJSON_Delete(saved); } if (!strcmp(kind, "edit") && (!resume || !*resume)) { char err[256]; if (cst_worktree_create(ctx->workspace_root, arg(c, "files"), j->worktree, sizeof(j->worktree), err, sizeof(err))) { cJSON_Delete(a); free(j); return out_text(out, 1, err); } cJSON_AddStringToObject(a, "_worktree", j->worktree); const char *files = arg(c, "files"); if (!files || strlen(files) >= sizeof(cfg->worker_scope)) { cJSON_Delete(a); free(j); return out_text(out, 1, "error: file scope is too large"); } cJSON_AddStringToObject(a, "_scope", files); } const char *steps = arg(c, "max_steps"); if (steps && *steps) cJSON_AddNumberToObject(a, "max_steps", atoi(steps)); char *json = cJSON_PrintUnformatted(a); cJSON_Delete(a); int parse = json ? cst_tool_call_from_native(&j->call, c->id, "task", json) : -1; free(json); if (parse) { worker_job_free(j); return out_text(out, 1, "error: invalid worker arguments"); } char id[32]; const char *resource = !strcmp(kind, "read") ? "" : j->worktree[0] ? j->worktree : "*"; char wt_copy[OS_PATH_MAX]; snprintf(wt_copy, sizeof(wt_copy), "%s", j->worktree); if (cst_job_start(cfg->runtime, "worker", resource, strcmp(kind, "external") != 0, worker_job_run, j, worker_job_free, id, sizeof(id))) { worker_job_free(j); return out_text(out, 1, "error: worker scheduler is full or paused"); } char message[OS_PATH_MAX + 128]; snprintf(message, sizeof(message), "job_id=%s%s%s; running asynchronously; continue independent work or wait for the result", id, wt_copy[0] ? "; worktree=" : "", wt_copy); return out_text(out, 0, message); } static int register_worker_tools(void) { const char *names[] = {"worker_start", "worker_diff", "worker_integrate"}; for (int i = 0; i < 3; i++) if (!cst_tool_find(names[i])) { cst_tool_def_t d = {0}; d.name = names[i]; d.owner = "builtin:worker-v2"; d.desc = i == 0 ? "Start an asynchronous typed worker. read uses Flash; edit uses Pro in an isolated Git worktree and requires files scope. external uses a protected serial resource. Use job_wait for results; small tasks should be done directly by the main agent." : i == 1 ? "Preview an isolated worker's scoped delta without changing the user's index." : "Integrate a reviewed worker delta. Conflicts preserve the worktree and patch; never overwrite user changes."; d.schema_json = i == 0 ? "{\"type\":\"object\",\"properties\":{\"kind\":{\"type\":\"string\",\"enum\":[\"read\",\"edit\",\"external\"]},\"prompt\":{\"type\":\"string\"},\"description\":{\"type\":\"string\"},\"files\":{\"type\":\"array\",\"items\":{\"type\":\"string\"}},\"worker_id\":{\"type\":\"string\"},\"max_steps\":{\"type\":\"integer\",\"minimum\":1}},\"required\":[\"kind\",\"prompt\"]}" : "{\"type\":\"object\",\"properties\":{\"worktree\":{\"type\":\"string\"}},\"required\":[\"worktree\"]}"; d.perm = i == 2 ? CST_TP_WRITE : i == 1 ? CST_TP_READ : CST_TP_ASK; d.run_ex = async_worker_tool; if (cst_tool_register(&d)) return -1; } return 0; } static int task_parallel(const ctl_tool_call_t *c, void *ud) { (void)ud; if (arg(c, "worker_id")) return 0; const char *want = arg(c, "agent"); if (!want || !want[0]) return 0; /* general */ cst_agent_role_t *v = calloc(CST_ROLES_MAX, sizeof(*v)); if (!v) return 0; int n = cst_roles_list(g_ws, v, CST_ROLES_MAX); int i = find_role(v, n, want); int ro = i >= 0 && v[i].read_only; cst_roles_free(v, n); free(v); return ro; } int cst_subagent_register_tool(const char *ws) { snprintf(g_ws, sizeof(g_ws), "%s", ws ? ws : ""); if (register_worker_tools()) return -1; if (cst_tool_find("task")) return 0; cst_agent_role_t *v = calloc(CST_ROLES_MAX, sizeof(*v)); if (!v) return -1; int n = cst_roles_list(g_ws, v, CST_ROLES_MAX); cst_buf_t names = {0}; for (int i = 0; i < n; i++) cst_buf_printf(&names, "%s%s", i ? ", " : "", v[i].name); cst_roles_free(v, n); free(v); cst_buf_t schema = {0}; cst_buf_printf( &schema, "{\"type\":\"object\",\"properties\":{" "\"description\":{\"type\":\"string\",\"description\":\"3-6 word " "summary shown to the user\"}," "\"prompt\":{\"type\":\"string\",\"description\":\"the complete task: " "goal, context, constraints and what to report back\"}," "\"worker_id\":{\"type\":\"string\",\"description\":\"resume a worker from this session; omit to create\"}," "\"max_steps\":{\"type\":\"integer\",\"minimum\":1,\"description\":\"request cap, at most the configured role limit\"}," "\"agent\":{\"type\":\"string\",\"description\":\"one of: %s " "(default general)\"}}," "\"required\":[\"description\",\"prompt\"]}", names.buf ? names.buf : "general"); cst_buf_free(&names); cst_tool_def_t d; memset(&d, 0, sizeof(d)); d.name = "task"; d.desc = "Hand a self-contained task to a sub-agent with its own context; " "you get back only its final report. Use it for broad searches " "or investigations that would flood your context, and for " "independent subtasks. It sees none of this conversation, so put " "everything it needs in prompt. Read-only agents (explore, plan) " "called together in one step run in parallel."; d.schema_json = schema.buf; d.perm = CST_TP_ASK; d.run_ex = task_run; d.parallel = task_parallel; d.owner = "builtin:task"; int rc = cst_tool_register(&d); cst_buf_free(&schema); return rc; }