commit bc1fdebe1d123a6a90bcb6e3d48f644a3f5c85c5 from: Isaac Meerleo date: Fri Jul 17 23:05:21 2026 UTC Integrate exact Compaction epochs commit - f3f30b82b058e173347141ecf1ccdebb3a6b2a16 commit + bc1fdebe1d123a6a90bcb6e3d48f644a3f5c85c5 blob - 270f362460e38a84a8a0639ff31e28dcc042408a blob + 2b3e8a927cc285b38200b0d8cee2a4588e33e1f1 --- src/fugu/Makefile +++ src/fugu/Makefile @@ -1,6 +1,7 @@ PROG= fugu SRCS= main.c coord.c priv.c conf.c parse.y \ - journal.c sysprompt.c tooldefs.c tool_display.c agentcfg.c skills.c output.c \ + journal.c system_epoch.c sysprompt.c tooldefs.c tool_display.c agentcfg.c \ + skills.c output.c \ turn_txn.c turn_mechanics.c anthropic_req.c claude_sub_req.c \ claude_sub_profile.c openai_req.c generation.c compaction.c msg.c \ json.c utf8.c model_window.c buf.c imsgev.c log.c xmalloc.c blob - e9856197ed488e1ee142c8e93a296cdf7922f96f blob + ce8376919a1290d045d04e1b91fd5519fc35e8c2 --- src/fugu/coord.c +++ src/fugu/coord.c @@ -56,6 +56,7 @@ #include "tooldefs.h" #include "tool_display.h" #include "agentcfg.h" +#include "system_epoch.h" #include "turn_mechanics.h" #include "turn_txn.h" #include "turn_txn_private.h" @@ -232,8 +233,18 @@ struct coord { /* project context (behavior.md 4) */ struct buf ctx_personal; /* ~/.fugu/FUGU.md bytes */ struct buf ctx_project; /* in-tree FUGU.md/AGENTS.md */ - struct buf compaction; /* /compact summary, if any */ - struct buf sysfull; /* system + injected context */ + struct system_epoch epoch; /* exact active System baseline */ + struct compaction_files *files; /* cumulative structured paths */ + struct buf compaction; /* complete continuation notes */ + struct buf compaction_prose; /* model-written portion only */ + struct system_epoch epoch_rollback; + struct compaction_files *files_rollback; + int have_epoch_rollback; + int have_context_rollback; + int64_t context_tokens_rollback; + int auto_compaction_pending; + size_t auto_compaction_retained_tokens; + int resume_restricted_notice; struct buf ctx_data; /* filled while wait == WAIT_CTX */ struct ctx_result ctx_res; int ctx_done; @@ -246,6 +257,18 @@ static void coord_teardown(struct coord *); static void agent_dispatch(int, short, void *); static void agent_wave_cleanup(struct coord *, int); static void agent_web_start_next(struct coord *); +static void epoch_rollback_discard(struct coord *); +static void epoch_rollback_restore(struct coord *); +static void auto_compaction_commit_notice(struct coord *); +static int epoch_reset_idle_gate(struct coord *, + const struct system_epoch *, const struct compaction_files *, size_t); +static int lead_tools(struct coord *, enum turn_tools_mode, + struct tool_def *, size_t); +static const char *validate_epoch_projection(struct coord *, + const struct system_epoch *, const struct msglist *, size_t, + enum turn_tools_mode, int); +static enum turn_io compact_projection(struct coord *, int, int, + enum turn_tools_mode, struct generation_usage *, char *, size_t); static struct subagent * agent_by_api_id(struct coord *c, uint32_t id) @@ -452,6 +475,20 @@ post_stream_test_delay(void) (void)nanosleep(&ts, NULL); } +static void +epoch_rollback_discard(struct coord *c) +{ + if (!c->have_epoch_rollback) + return; + system_epoch_free(&c->epoch_rollback); + system_epoch_init(&c->epoch_rollback); + compaction_files_free(c->files_rollback); + c->files_rollback = NULL; + c->have_epoch_rollback = 0; + c->have_context_rollback = 0; + c->context_tokens_rollback = 0; +} + enum commit_gate { COMMIT_GATE_COMMITTED, COMMIT_GATE_CANCELLED, @@ -480,8 +517,11 @@ commit_turn(struct coord *c, enum turn_txn_commit outc gate = COMMIT_GATE_QUIT; else if (c->want_switch) gate = COMMIT_GATE_SWITCH; - else + else { turn_txn_commit(c->turn, outcome); + auto_compaction_commit_notice(c); + epoch_rollback_discard(c); + } return (gate); } sigemptyset(&block); @@ -498,8 +538,11 @@ commit_turn(struct coord *c, enum turn_txn_commit outc gate = COMMIT_GATE_QUIT; else if (c->want_switch) gate = COMMIT_GATE_SWITCH; - else + else { turn_txn_commit(c->turn, outcome); + auto_compaction_commit_notice(c); + epoch_rollback_discard(c); + } if (sigprocmask(SIG_SETMASK, &oldmask, NULL) == -1) fatal("sigprocmask restore SIGINT"); return (gate); @@ -1089,7 +1132,55 @@ emit_msg(struct coord *c, int uitype, FILE *fallback, static int64_t effective_window(struct coord *); /* defined below */ +static void +auto_compaction_commit_notice(struct coord *c) +{ + if (!c->auto_compaction_pending) + return; + emit_note(c, "conversation compacted automatically " + "(~%lld tokens retained)", + (long long)c->auto_compaction_retained_tokens); + c->auto_compaction_pending = 0; + c->auto_compaction_retained_tokens = 0; +} + +/* + * The final idle reset gate mirrors the Turn commit boundary: drain decisions, + * block SIGINT, check already-pending signals, then append+sync+publish while + * the boundary is closed. A decision arriving afterward belongs to the new + * epoch. Returns 0 committed, 1 interrupted, or -1 write/validation failure. + */ static int +epoch_reset_idle_gate(struct coord *c, const struct system_epoch *epoch, + const struct compaction_files *files, size_t retain_first) +{ + sigset_t block, oldmask, pending; + int interrupted, rc; + + (void)event_loop(EVLOOP_NONBLOCK); + if (!c->have_sigint_ev) { + if (c->cancelled || c->want_quit || c->want_switch) + return (1); + return (turn_txn_epoch_reset(c->turn, epoch, files, + retain_first)); + } + sigemptyset(&block); + sigaddset(&block, SIGINT); + if (sigprocmask(SIG_BLOCK, &block, &oldmask) == -1) + fatal("sigprocmask SIGINT"); + (void)event_loop(EVLOOP_NONBLOCK); + if (sigpending(&pending) == -1) + fatal("sigpending"); + interrupted = c->cancelled || c->want_quit || c->want_switch || + sigismember(&pending, SIGINT); + rc = interrupted ? 1 : turn_txn_epoch_reset(c->turn, epoch, files, + retain_first); + if (sigprocmask(SIG_SETMASK, &oldmask, NULL) == -1) + fatal("sigprocmask restore SIGINT"); + return (rc); +} + +static int usage_add(struct a_usage *dst, const struct a_usage *src) { if (src->input_tokens < 0 || src->output_tokens < 0 || @@ -1359,10 +1450,7 @@ stream_event(struct coord *c, struct imsg *imsg) au.cache_write = ev.usage.cache_write; if (!c->turn_data_exceeded && !c->turn_usage_exceeded) { if (lead_usage_account(c, &au) == -1) { - if (c->turn_generation_io != NULL) - c->turn_usage_exceeded = 1; - else - c->turn_data_exceeded = 1; + c->turn_usage_exceeded = 1; } else machine_usage(c, &au); @@ -2665,58 +2753,164 @@ ctx_read_project(struct coord *c, const char *path) } } -/* - * Compose the base system prompt with the injected context (personal - * first, then project, behavior.md 4), rebuilt whenever the context - * changes. sysfull is always valid after this call. - */ +static const struct buf * +lead_system(const struct coord *c) +{ + const struct buf *selected; + + selected = system_epoch_select(&c->epoch, c->conf->project_context); + if (selected == NULL || c->epoch.version != SYSTEM_EPOCH_VERSION) + fatalx("System epoch is not initialized"); + return (selected); +} + +static int +build_system_epoch(struct coord *c, struct system_epoch *epoch, + enum system_epoch_reason reason, const struct buf *personal, + const struct buf *project, const struct buf *summary) +{ + const char *base; + struct system_epoch_input input; + + base = c->conf->system != NULL ? c->conf->system : fugu_system_prompt; + memset(&input, 0, sizeof(input)); + input.reason = reason; + input.base.data = base; + input.base.len = strlen(base); + input.personal.data = personal->data; + input.personal.len = personal->len; + input.project.data = project->data; + input.project.len = project->len; + input.summary.data = summary->data; + input.summary.len = summary->len; + return (system_epoch_build(epoch, &input)); +} + static void -rebuild_system(struct coord *c) +mirror_system_epoch(struct coord *c) { - const char *base = c->conf->system != NULL ? c->conf->system : - fugu_system_prompt; + buf_reset(&c->ctx_personal); + buf_reset(&c->ctx_project); + buf_reset(&c->compaction); + buf_reset(&c->compaction_prose); + if (c->epoch.personal.len > 0) + buf_add(&c->ctx_personal, c->epoch.personal.data, + c->epoch.personal.len); + if (c->epoch.project.len > 0) + buf_add(&c->ctx_project, c->epoch.project.data, + c->epoch.project.len); + if (c->epoch.summary.len > 0) + buf_add(&c->compaction, c->epoch.summary.data, + c->epoch.summary.len); + if (compaction_summary_prose(&c->compaction_prose, + c->epoch.summary.data, c->epoch.summary.len, c->files) == -1) + fatalx("cannot restore Compaction prose"); +} - buf_reset(&c->sysfull); - buf_addstr(&c->sysfull, base); - if (c->ctx_personal.len > 0) { - buf_addstr(&c->sysfull, "\n\n"); - buf_add(&c->sysfull, c->ctx_personal.data, c->ctx_personal.len); +static struct compaction_files * +compaction_files_copy(const struct compaction_files *source) +{ + struct compaction_files *copy; + + copy = compaction_files_new(); + if (compaction_files_merge(copy, source) == -1) { + compaction_files_free(copy); + return (NULL); } - if (c->ctx_project.len > 0) { - buf_addstr(&c->sysfull, "\n\n"); - buf_add(&c->sysfull, c->ctx_project.data, c->ctx_project.len); + return (copy); +} + +static int +epoch_rollback_begin(struct coord *c) +{ + if (c->have_epoch_rollback) + return (0); + if (system_epoch_copy(&c->epoch_rollback, &c->epoch) == -1) + return (-1); + c->files_rollback = compaction_files_copy(c->files); + if (c->files_rollback == NULL) { + system_epoch_free(&c->epoch_rollback); + system_epoch_init(&c->epoch_rollback); + return (-1); } - if (c->compaction.len > 0) { - buf_addstr(&c->sysfull, "\n\nSummary of the conversation so " - "far (your own continuation notes):\n\n"); - buf_add(&c->sysfull, c->compaction.data, c->compaction.len); - } - buf_addc(&c->sysfull, '\0'); /* NUL-terminate for strlen safety */ - c->sysfull.len--; /* but do not count it */ + c->have_context_rollback = c->have_context; + c->context_tokens_rollback = c->context_tokens; + c->have_epoch_rollback = 1; + return (0); } +static void +epoch_rollback_restore(struct coord *c) +{ + if (!c->have_epoch_rollback) + return; + system_epoch_free(&c->epoch); + c->epoch = c->epoch_rollback; + system_epoch_init(&c->epoch_rollback); + compaction_files_free(c->files); + c->files = c->files_rollback; + c->files_rollback = NULL; + c->have_epoch_rollback = 0; + mirror_system_epoch(c); + c->have_context = c->have_context_rollback; + c->context_tokens = c->context_tokens_rollback; + c->have_context_rollback = 0; + c->context_tokens_rollback = 0; + c->auto_compaction_pending = 0; + c->auto_compaction_retained_tokens = 0; +} + /* * Before the steady pledge (while rpath and the journal are still * reachable): a resumed session replays its journaled context, a * fresh session reads the personal file (behavior.md 4, I14). */ static void -context_pre_pledge(struct coord *c, int resumed, const char *sessdir, - const char *resume_id) +context_pre_pledge(struct coord *c, int resumed, + struct journal_state *restored) { if (resumed) { - /* - * Always recover the compaction summary -- it is - * conversation state, not a context file, and /compact is - * not gated on project_context (I14). The project/personal - * files apply only when project_context is on. - */ - if (journal_replay_context(sessdir, resume_id, - &c->ctx_personal, &c->ctx_project, &c->compaction) == -1) - fatalx("cannot replay context for session %s", resume_id); - if (!c->conf->project_context) { + if (restored == NULL) + fatalx("missing restored Journal state"); + if (restored->have_epoch) { + system_epoch_free(&c->epoch); + c->epoch = restored->epoch; + system_epoch_init(&restored->epoch); + compaction_files_free(c->files); + c->files = restored->files; + restored->files = NULL; + mirror_system_epoch(c); + c->resume_restricted_notice = + !c->conf->project_context && + (c->epoch.personal.len > 0 || + c->epoch.project.len > 0); + } else { buf_reset(&c->ctx_personal); buf_reset(&c->ctx_project); + buf_reset(&c->compaction); + buf_reset(&c->compaction_prose); + if (restored->legacy_personal.len > 0) + buf_add(&c->ctx_personal, + restored->legacy_personal.data, + restored->legacy_personal.len); + if (restored->legacy_project.len > 0) + buf_add(&c->ctx_project, + restored->legacy_project.data, + restored->legacy_project.len); + if (restored->legacy_summary.len > 0) { + buf_add(&c->compaction, + restored->legacy_summary.data, + restored->legacy_summary.len); + buf_add(&c->compaction_prose, + restored->legacy_summary.data, + restored->legacy_summary.len); + } + /* Preserve the old reconstruction rule for the one-time + * migration boundary. */ + if (!c->conf->project_context) { + buf_reset(&c->ctx_personal); + buf_reset(&c->ctx_project); + } } } else if (c->conf->project_context) read_personal(c); @@ -2728,28 +2922,49 @@ context_pre_pledge(struct coord *c, int resumed, const * composed. A resumed session already has its context from replay. */ static void -context_post_bringup(struct coord *c, int resumed) +context_post_bringup(struct coord *c, int resumed, const struct msglist *seed) { + struct system_epoch next; + const char *why; + if (c->conf->project_context && !resumed) { ctx_read_project(c, "FUGU.md"); if (c->ctx_project.len == 0) ctx_read_project(c, "AGENTS.md"); - journal_context(c->journal, - (const char *)c->ctx_personal.data, c->ctx_personal.len, - (const char *)c->ctx_project.data, c->ctx_project.len); + } + if (c->epoch.version == 0) { + system_epoch_init(&next); + if (build_system_epoch(c, &next, + resumed ? SYSTEM_EPOCH_MIGRATION : SYSTEM_EPOCH_START, + &c->ctx_personal, &c->ctx_project, &c->compaction) == -1) + fatalx("cannot build initial System epoch"); + why = validate_epoch_projection(c, &next, seed, 0, + TURN_TOOLS_ENABLED, 0); + if (why != NULL || + journal_system_epoch(c->journal, &next, c->files, NULL, + 0) == -1) + fatalx("cannot install initial System epoch: %s", + why != NULL ? why : "Journal write failed"); journal_sync(c->journal); - } - rebuild_system(c); + system_epoch_free(&c->epoch); + c->epoch = next; + system_epoch_init(&next); + system_epoch_free(&next); + } else if ((why = validate_epoch_projection(c, &c->epoch, seed, 0, + TURN_TOOLS_ENABLED, 0)) != NULL) + fatalx("restored System epoch is not Provider-valid: %s", why); + if (c->resume_restricted_notice) + emit_note(c, "project context disabled for resumed session"); } /* Translate a borrowed Projection into the active wire dialect. */ -static void -build_provider_request(struct coord *c, struct buf *body, +static int +build_provider_request(struct coord *c, int provider_type, struct buf *body, const char *system, size_t systemlen, const struct msglist *messages, const struct msg *suffix, const struct tool_def *tools, int ntools, int cache_tail_msg) { - if (c->conf->provider_type == PROVIDER_OPENAI) { + if (provider_type == PROVIDER_OPENAI) { struct openai_req req; memset(&req, 0, sizeof(req)); @@ -2762,7 +2977,8 @@ build_provider_request(struct coord *c, struct buf *bo req.conv = messages; req.suffix = suffix; openai_build_request(body, &req); - } else if (c->conf->provider_type == PROVIDER_CLAUDE) { + return (0); + } else if (provider_type == PROVIDER_CLAUDE) { struct claude_sub_req req; memset(&req, 0, sizeof(req)); @@ -2776,9 +2992,8 @@ build_provider_request(struct coord *c, struct buf *bo req.suffix = suffix; req.cache = 1; req.cache_tail_msg = cache_tail_msg; - if (claude_sub_build_request(body, &req) == -1) - fatalx("cannot build Claude subscription request"); - } else { + return (claude_sub_build_request(body, &req)); + } else if (provider_type == PROVIDER_ANTHROPIC) { struct anthropic_req req; memset(&req, 0, sizeof(req)); @@ -2793,7 +3008,9 @@ build_provider_request(struct coord *c, struct buf *bo req.cache = 1; req.cache_tail_msg = cache_tail_msg; anthropic_build_request(body, &req); + return (0); } + return (-1); } static void @@ -2852,40 +3069,248 @@ lead_checkpoint(void *arg) return (TURN_CHECKPOINT_CONTINUE); } +static int +lead_tools(struct coord *c, enum turn_tools_mode mode, struct tool_def *tools, + size_t capacity) +{ + const struct tool_def *base, *agent; + int nbase, ntools = 0; + + if (mode == TURN_TOOLS_DISABLED) + return (0); + if (mode != TURN_TOOLS_ENABLED) + return (-1); + base = tooldefs(c->conf, c->skills, &nbase); + if (nbase < 0 || (size_t)nbase > capacity) + return (-1); + memcpy(tools, base, (size_t)nbase * sizeof(tools[0])); + ntools = nbase; + agent = agentcfg_tool(c->agentcfg, c->active_provider_slot, + c->conf->provider_type, c->conf->model); + if (agent != NULL) { + if ((size_t)ntools == capacity) + return (-1); + tools[ntools++] = *agent; + } + return (ntools); +} + +/* + * Make a borrowed, forward-only TAILQ view of one Projection suffix. Provider + * codecs only iterate conversation heads and never mutate them; the original + * message links therefore remain the single canonical ownership structure. + */ +static int +projection_suffix_view(const struct msglist *messages, size_t retain_first, + struct msglist *view) +{ + const struct msg *message; + size_t index = 0; + + if (messages == NULL || view == NULL) + return (-1); + TAILQ_INIT(view); + TAILQ_FOREACH(message, messages, entry) { + if (index == retain_first) { + view->tqh_first = (struct msg *)message; + view->tqh_last = messages->tqh_last; + return (0); + } + index++; + } + return (index == retain_first ? 0 : -1); +} + +/* + * Validate an epoch boundary against the exact live Provider codec before it + * becomes durable. Automatic cuts additionally have to fall back below the + * strict conservative threshold; otherwise the next prepare hook would only + * compact the same state again. + */ +static const char * +validate_epoch_projection(struct coord *c, const struct system_epoch *epoch, + const struct msglist *messages, size_t retain_first, + enum turn_tools_mode mode, int check_threshold) +{ + static const int providers[] = { + PROVIDER_ANTHROPIC, PROVIDER_OPENAI, PROVIDER_CLAUDE + }; + const struct buf *system, *variants[2]; + const struct msg *message; + struct tool_def tools[20]; + struct msglist retained; + struct msg *placeholder = NULL; + struct buf body; + int64_t window; + size_t estimate; + int have_user = 0, i, ntools, threshold; + size_t pi; + + if (epoch == NULL || messages == NULL || + (system = system_epoch_select(epoch, + c->conf->project_context)) == NULL) + return ("invalid policy-selected System baseline"); + ntools = lead_tools(c, mode, tools, sizeof(tools) / sizeof(tools[0])); + if (ntools == -1) + return ("invalid prospective Tool catalogue"); + if (projection_suffix_view(messages, retain_first, &retained) == -1) + return ("invalid retained Projection cut"); + /* An idle reset has no complete Provider request until the next Owner + * message. Use the smallest valid future User shell solely to exercise + * provider-specific System lowering and JSON expansion. */ + TAILQ_FOREACH(message, &retained, entry) + if (message->role == ROLE_USER) { + have_user = 1; + break; + } + if (!have_user && check_threshold) + return ("active retained Projection has no User message"); + if (!have_user) { + placeholder = msg_new(ROLE_USER); + msg_add_text(placeholder, "", 0); + } + variants[0] = &epoch->full; + variants[1] = &epoch->restricted; + for (i = 0; i < 2; i++) { + for (pi = 0; pi < sizeof(providers) / sizeof(providers[0]); pi++) { + buf_init(&body); + if (build_provider_request(c, providers[pi], &body, + (const char *)variants[i]->data, variants[i]->len, + &retained, placeholder, ntools > 0 ? tools : NULL, + ntools, -1) == -1) { + if (placeholder != NULL) + msg_free(placeholder); + buf_free(&body); + return ("stored System variant is rejected by a Provider codec"); + } + if (body.len > FUGU_REQUEST_MAX) { + if (placeholder != NULL) + msg_free(placeholder); + buf_free(&body); + return ("stored System variant exceeds a Provider wire bound"); + } + buf_free(&body); + } + } + if (placeholder != NULL) + msg_free(placeholder); + if (!check_threshold || (window = effective_window(c)) == 0) + return (NULL); + if (compaction_estimate_request(system->data, system->len, &retained, + ntools > 0 ? tools : NULL, (size_t)ntools, &estimate) == -1) + return ("prospective context estimate exceeded its bound"); + threshold = compaction_threshold_exceeded(estimate, (size_t)window, + (size_t)c->conf->compact_reserve_tokens); + if (threshold == -1) + return ("Compaction reserve is not below the context window"); + if (threshold != 0) + return ("replacement Projection remains above the Compaction threshold"); + return (NULL); +} + +static enum turn_io +lead_prepare_generation(void *arg, enum turn_tools_mode mode, + struct generation_usage *usage, char *error, size_t errorsz) +{ + struct coord *c = arg; + const struct buf *system; + struct tool_def tools[20]; + struct turn_txn_projection projection; + enum turn_io status; + int64_t window; + size_t estimate; + int ntools, threshold; + + memset(usage, 0, sizeof(*usage)); + window = effective_window(c); + if (!c->conf->auto_compact || window == 0) + return (TURN_IO_OK); + ntools = lead_tools(c, mode, tools, + sizeof(tools) / sizeof(tools[0])); + if (ntools == -1) { + (void)snprintf(error, errorsz, "%s", + "cannot estimate the prospective Tool catalogue"); + return (TURN_IO_ERROR); + } + turn_txn_projection(c->turn, &projection); + system = lead_system(c); + if (compaction_estimate_request(system->data, system->len, + projection.messages, ntools > 0 ? tools : NULL, (size_t)ntools, + &estimate) == -1) { + (void)snprintf(error, errorsz, "%s", + "prospective context estimate exceeded its bound"); + return (TURN_IO_ERROR); + } + threshold = compaction_threshold_exceeded(estimate, (size_t)window, + (size_t)c->conf->compact_reserve_tokens); + if (threshold == -1) { + (void)snprintf(error, errorsz, + "compact_reserve_tokens (%lld) must be less than the " + "effective context window (%lld)", + (long long)c->conf->compact_reserve_tokens, + (long long)window); + return (TURN_IO_ERROR); + } + if (threshold == 0) + return (TURN_IO_OK); + status = compact_projection(c, 1, 0, mode, usage, error, errorsz); + if (status == TURN_IO_OK) + send_status(c, UI_BUSY); + return (status); +} + +static enum turn_io +lead_recover_context_overflow(void *arg, enum turn_tools_mode mode, + struct generation_usage *usage, char *error, size_t errorsz) +{ + struct coord *c = arg; + enum turn_io status; + + memset(usage, 0, sizeof(*usage)); + if (!c->conf->auto_compact) { + (void)snprintf(error, errorsz, "%s", + "automatic Compaction is disabled"); + return (TURN_IO_ERROR); + } + status = compact_projection(c, 1, 0, mode, usage, error, errorsz); + if (status == TURN_IO_OK) { + emit_note(c, "retrying rejected Generation after context overflow"); + send_status(c, UI_RETRY); + } + return (status); +} + /* Build, send, and collect one provider-neutral Lead Generation. */ static enum turn_io lead_generate(void *arg, const struct turn_generation_request *request, struct turn_generation_io *io, char *error, size_t errorsz) { struct coord *c = arg; - const struct tool_def *base, *agent; + const struct buf *system; struct tool_def tools[20]; struct buf body; - int nbase, ntools = 0, rc; + int ntools, rc; buf_init(&body); - if (request->tools == TURN_TOOLS_ENABLED) { - base = tooldefs(c->conf, c->skills, &nbase); - if (nbase > (int)(sizeof(tools) / sizeof(tools[0]))) - fatalx("too many tool definitions"); - memcpy(tools, base, (size_t)nbase * sizeof(tools[0])); - ntools = nbase; - agent = agentcfg_tool(c->agentcfg, c->active_provider_slot, - c->conf->provider_type, c->conf->model); - if (agent != NULL) { - if (ntools >= (int)(sizeof(tools) / sizeof(tools[0]))) - fatalx("too many tool definitions"); - tools[ntools++] = *agent; - } - } else if (request->tools != TURN_TOOLS_DISABLED) { + ntools = lead_tools(c, request->tools, tools, + sizeof(tools) / sizeof(tools[0])); + if (ntools == -1) { buf_free(&body); - (void)snprintf(error, errorsz, "%s", "invalid Lead tool mode"); + (void)snprintf(error, errorsz, "%s", + "invalid Lead tool catalogue"); return (TURN_IO_ERROR); } - build_provider_request(c, &body, (const char *)c->sysfull.data, - c->sysfull.len, request->projection.messages, NULL, + system = lead_system(c); + if (build_provider_request(c, c->conf->provider_type, &body, + (const char *)system->data, + system->len, request->projection.messages, NULL, ntools > 0 ? tools : NULL, ntools, - request->projection.cache_tail_msg); + request->projection.cache_tail_msg) == -1) { + buf_free(&body); + (void)snprintf(error, errorsz, "%s", + "Provider rejected the canonical request shape"); + return (TURN_IO_ERROR); + } if (body.len > FUGU_REQUEST_MAX) { size_t mb = body.len / (1024 * 1024); @@ -3230,8 +3655,6 @@ bad: static void agent_system(struct coord *c, struct buf *out) { - const char *base = c->conf->system != NULL ? c->conf->system : - fugu_system_prompt; static const char guidance[] = "\n\nYou are an ephemeral read-only subagent. Work only on the " "self-contained task you were given. Explore and verify with the " @@ -3240,14 +3663,14 @@ agent_system(struct coord *c, struct buf *out) "useful file:line references. Do not address the Owner directly."; buf_reset(out); - buf_addstr(out, base); - if (c->ctx_personal.len > 0) { + buf_add(out, c->epoch.base.data, c->epoch.base.len); + if (c->conf->project_context && c->epoch.personal.len > 0) { buf_addstr(out, "\n\n"); - buf_add(out, c->ctx_personal.data, c->ctx_personal.len); + buf_add(out, c->epoch.personal.data, c->epoch.personal.len); } - if (c->ctx_project.len > 0) { + if (c->conf->project_context && c->epoch.project.len > 0) { buf_addstr(out, "\n\n"); - buf_add(out, c->ctx_project.data, c->ctx_project.len); + buf_add(out, c->epoch.project.data, c->epoch.project.len); } buf_add(out, guidance, sizeof(guidance) - 1); } @@ -3587,6 +4010,9 @@ static int abandon_turn(struct coord *c, int status) { turn_txn_abandon(c->turn); + epoch_rollback_restore(c); + c->auto_compaction_pending = 0; + c->auto_compaction_retained_tokens = 0; reset_tcalls(c); c->turn_generation_io = NULL; c->turn_tool_results = NULL; @@ -3667,6 +4093,11 @@ run_turn(struct coord *c, const char *prompt, size_t p adapter.accept = lead_accept; adapter.assistant_accepted = lead_assistant_accepted; adapter.checkpoint = lead_checkpoint; + if (c->conf->auto_compact) { + adapter.prepare_generation = lead_prepare_generation; + adapter.recover_context_overflow = + lead_recover_context_overflow; + } adapter.generate = lead_generate; adapter.execute = lead_execute; memset(&options, 0, sizeof(options)); @@ -3751,11 +4182,16 @@ run_turn(struct coord *c, const char *prompt, size_t p static int64_t est_tokens(size_t); /* defined below */ /* /clear: reset the active context; the next message starts fresh. */ -static void +static int cmd_clear(struct coord *c) { - struct turn_txn_context context; - struct turn_txn_context *contextp = NULL; + struct compaction_files *files; + struct system_epoch epoch; + struct turn_txn_projection projection; + struct buf empty; + const char *why = NULL; + const char *old_ctx_file = c->ctx_file; + int gate = -1, ok = 0; /* the in-tree file is re-read after a clear (behavior.md 4) */ if (c->conf->project_context) { @@ -3765,125 +4201,272 @@ cmd_clear(struct coord *c) ctx_read_project(c, f); if (c->ctx_project.len == 0 && strcmp(f, "AGENTS.md") != 0) ctx_read_project(c, "AGENTS.md"); - context.personal = (const char *)c->ctx_personal.data; - context.personal_len = c->ctx_personal.len; - context.project = (const char *)c->ctx_project.data; - context.project_len = c->ctx_project.len; - contextp = &context; + } else + buf_reset(&c->ctx_project); + system_epoch_init(&epoch); + buf_init(&empty); + files = compaction_files_new(); + turn_txn_projection(c->turn, &projection); + if (build_system_epoch(c, &epoch, SYSTEM_EPOCH_CLEAR, + &c->ctx_personal, &c->ctx_project, &empty) == -1) + why = "replacement System epoch exceeds its bound"; + else if ((why = validate_epoch_projection(c, &epoch, + projection.messages, projection.message_count, TURN_TOOLS_ENABLED, + 0)) == NULL) { + gate = epoch_reset_idle_gate(c, &epoch, files, + projection.message_count); + if (gate == 0) + ok = 1; + else if (gate == 1) + why = "clear interrupted before its commit boundary"; + else + why = "Journal reset failed"; } - turn_txn_clear(c->turn, contextp); - buf_reset(&c->compaction); /* a reset voids the summary */ - rebuild_system(c); /* drop the summary either way */ + buf_free(&empty); + if (!ok) { + system_epoch_free(&epoch); + compaction_files_free(files); + mirror_system_epoch(c); + c->ctx_file = old_ctx_file; + emit_error(c, "fugu: could not install clear System epoch%s%s", + why != NULL ? ": " : "", why != NULL ? why : ""); + return (gate == 1 ? 3 : 1); + } + system_epoch_free(&c->epoch); + c->epoch = epoch; + system_epoch_init(&epoch); + compaction_files_free(c->files); + c->files = files; + mirror_system_epoch(c); context_invalidate(c); if (c->ui) ui_send(c, FUGU_IMSG_UI_CLEARED, NULL, 0); emit_note(c, "context cleared"); + return (0); } -/* - * /compact (behavior.md 3): one extra API call (no tools) summarizes - * the conversation into continuation notes, which then replace the - * active context. The summary is injected like the project context - * and journaled so a resume carries the compacted state. - */ -static int -cmd_compact(struct coord *c) +/* One shared manual/automatic Compaction operation. */ +static enum turn_io +compact_projection(struct coord *c, int active, int force, + enum turn_tools_mode mode, struct generation_usage *usage, char *error, + size_t errorsz) { + struct compaction_cut_plan plan; + struct compaction_files *files = NULL; struct generation_result generation; - struct generation_span text; + struct generation_span text; + struct system_epoch epoch; + struct system_epoch_input input; struct turn_txn_projection projection; - struct msglist compact_messages; - struct msg *document_message; - struct buf body, document; - const char *why; - int pump_rc; + struct msglist compact_messages; + struct msg *document_message; + struct buf body, document, summary; + const char *why; + int began_rollback = 0, pump_rc, reset_rc; - turn_txn_projection(c->turn, &projection); - if (projection.message_count == 0) { - emit_error(c, "fugu: nothing to compact"); - return (1); - } - - send_status(c, UI_SUMMARIZE); /* distinct busy state (2.1) */ - - /* Quote the Projection as data under a summarizer-only instruction. */ + if (usage != NULL) + memset(usage, 0, sizeof(*usage)); + if (error != NULL && errorsz > 0) + error[0] = '\0'; + memset(&generation, 0, sizeof(generation)); + memset(&plan, 0, sizeof(plan)); + system_epoch_init(&epoch); + buf_init(&body); buf_init(&document); - if (compaction_serialize(&document, projection.messages, - c->compaction.data, c->compaction.len, FUGU_REQUEST_MAX) == -1) { - buf_free(&document); - emit_error(c, "fugu: too large to compact"); - return (1); - } + buf_init(&summary); TAILQ_INIT(&compact_messages); + turn_txn_projection(c->turn, &projection); + if (projection.message_count == 0) { + why = "nothing to compact"; + goto fail; + } + if (compaction_plan_cut(projection.messages, + (size_t)c->conf->compact_keep_tokens, active, force, &plan) == -1 || + plan.history_prefix > plan.retain_first || + plan.current_turn_prefix != + plan.retain_first - plan.history_prefix) { + why = "could not select a safe Compaction cut"; + goto fail; + } + if (plan.retain_first == 0) { + why = "no safe Projection prefix can be compacted"; + goto fail; + } + files = compaction_files_copy(c->files); + if (files == NULL || compaction_files_collect_prefix(files, + projection.messages, plan.retain_first) == -1) { + why = "structured file state exceeds its bound"; + goto fail; + } + if (compaction_serialize_cut(&document, projection.messages, + plan.history_prefix, plan.current_turn_prefix, + c->compaction_prose.data, c->compaction_prose.len, + FUGU_REQUEST_MAX) == -1) { + why = "Compaction source exceeds the request bound"; + goto fail; + } + + send_status(c, UI_SUMMARIZE); document_message = msg_new(ROLE_USER); msg_add_text(document_message, (const char *)document.data, document.len); TAILQ_INSERT_TAIL(&compact_messages, document_message, entry); - buf_init(&body); - build_provider_request(c, &body, compaction_system_prompt, + if (build_provider_request(c, c->conf->provider_type, &body, + compaction_system_prompt, strlen(compaction_system_prompt), &compact_messages, NULL, NULL, 0, - -1); /* no tools (S3) */ + -1) == -1) { + msglist_free(&compact_messages); + why = "Provider rejected the Compaction request shape"; + goto fail; + } msglist_free(&compact_messages); - buf_free(&document); if (body.len > FUGU_REQUEST_MAX) { - buf_free(&body); - emit_error(c, "fugu: too large to compact"); - return (1); + why = "serialized Compaction request exceeds its bound"; + goto fail; } send_request(c, &body, c->lead_api_id, c->active_provider_slot); - buf_free(&body); - c->turn_data = 0; - c->turn_data_exceeded = 0; - c->turn_usage_exceeded = 0; - pump_rc = pump_generation(c, FUGU_TURN_DATA_MAX, 1, &generation); - - if (cancel_checkpoint(c)) { + pump_rc = pump_generation(c, FUGU_TURN_DATA_MAX, 0, &generation); + if (usage != NULL) { + usage->input_tokens = c->lead_usage.input_tokens; + usage->output_tokens = c->lead_usage.output_tokens; + usage->cache_read = c->lead_usage.cache_read; + usage->cache_write = c->lead_usage.cache_write; + } + if (cancel_checkpoint(c) || c->want_quit || c->want_switch) { generation_result_dispose(&generation); - return (3); + compaction_files_free(files); + buf_free(&summary); + buf_free(&document); + buf_free(&body); + system_epoch_free(&epoch); + return (TURN_IO_INTERRUPTED); } if (pump_rc == -1) { - generation_result_dispose(&generation); - emit_error(c, "fugu: compaction failed: provider stream ended " - "before its terminal"); - return (1); + why = "Provider stream ended before the Compaction terminal"; + goto generation_fail; } - c->turn_data += generation.data_used; - if (generation.outcome == GENERATION_INVALID && - generation.invalid_cause == GENERATION_INVALID_DATA_BOUND) - c->turn_data_exceeded = 1; + if (c->turn_usage_exceeded) { + why = "Compaction usage exceeded the Turn usage bound"; + goto generation_fail; + } memset(&text, 0, sizeof(text)); if (generation.outcome == GENERATION_NORMAL && generation.invalid_cause == GENERATION_INVALID_NONE && generation.assistant != NULL && generation.tool_calls == 0) text = lead_assistant_text(generation.assistant); - if (c->turn_data_exceeded || generation.outcome != GENERATION_NORMAL || + if (generation.outcome != GENERATION_NORMAL || generation.invalid_cause != GENERATION_INVALID_NONE || generation.assistant == NULL || generation.tool_calls != 0 || text.len == 0) { why = c->err[0] != '\0' ? c->err : - "summary was incomplete or contained tool calls"; - generation_result_dispose(&generation); - emit_error(c, "fugu: compaction failed: %s", why); - return (1); + "summary was incomplete or contained Tool calls"; + goto generation_fail; } - if (cancel_checkpoint(c)) { - generation_result_dispose(&generation); - return (3); + buf_add(&summary, text.data, text.len); + if (compaction_files_append(&summary, files, + SYSTEM_EPOCH_SUMMARY_MAX) == -1) { + why = "replacement continuation notes exceed their bound"; + goto generation_fail; } - buf_reset(&c->text); - buf_add(&c->text, text.data, text.len); generation_result_dispose(&generation); - /* The durable summary replaces the Projection and becomes context. */ - turn_txn_compact(c->turn, c->text.data, c->text.len); - buf_reset(&c->compaction); - buf_add(&c->compaction, c->text.data, c->text.len); - rebuild_system(c); + memset(&input, 0, sizeof(input)); + input.reason = SYSTEM_EPOCH_COMPACT; + input.base.data = c->epoch.base.data; + input.base.len = c->epoch.base.len; + input.personal.data = c->epoch.personal.data; + input.personal.len = c->epoch.personal.len; + input.project.data = c->epoch.project.data; + input.project.len = c->epoch.project.len; + input.summary.data = summary.data; + input.summary.len = summary.len; + if (system_epoch_build(&epoch, &input) == -1) { + why = "replacement System epoch exceeds its bound"; + goto fail; + } + if ((why = validate_epoch_projection(c, &epoch, projection.messages, + plan.retain_first, mode, !force)) != NULL) + goto fail; + if (active && !c->have_epoch_rollback) { + if (epoch_rollback_begin(c) == -1) { + why = "could not preserve the active Turn epoch"; + goto fail; + } + began_rollback = 1; + } + reset_rc = active ? turn_txn_epoch_reset(c->turn, &epoch, files, + plan.retain_first) : epoch_reset_idle_gate(c, &epoch, files, + plan.retain_first); + if (reset_rc == 1) { + compaction_files_free(files); + buf_free(&summary); + buf_free(&document); + buf_free(&body); + system_epoch_free(&epoch); + return (TURN_IO_INTERRUPTED); + } + if (reset_rc == -1) { + if (began_rollback) + epoch_rollback_discard(c); + why = "could not synchronize the replacement System epoch"; + goto fail; + } + system_epoch_free(&c->epoch); + c->epoch = epoch; + system_epoch_init(&epoch); + compaction_files_free(c->files); + c->files = files; + files = NULL; + mirror_system_epoch(c); context_invalidate(c); - if (c->ui) + if (force && c->ui) ui_send(c, FUGU_IMSG_UI_CLEARED, NULL, 0); - emit_note(c, "conversation compacted (~%lld tokens of notes)", - (long long)est_tokens(c->compaction.len)); + if (force) + emit_note(c, "conversation compacted (~%lld tokens of notes)", + (long long)est_tokens(c->compaction.len)); + else { + c->auto_compaction_pending = 1; + c->auto_compaction_retained_tokens = plan.retained_tokens; + } + buf_free(&summary); + buf_free(&document); + buf_free(&body); + system_epoch_free(&epoch); + return (TURN_IO_OK); + +generation_fail: + generation_result_dispose(&generation); +fail: + if (error != NULL && errorsz > 0) + (void)snprintf(error, errorsz, "%s", why); + msglist_free(&compact_messages); + compaction_files_free(files); + buf_free(&summary); + buf_free(&document); + buf_free(&body); + system_epoch_free(&epoch); + return (TURN_IO_ERROR); +} + +static int +cmd_compact(struct coord *c) +{ + char error[TURN_ERROR_MAX]; + enum turn_io status; + + memset(&c->turn_usage, 0, sizeof(c->turn_usage)); + c->turn_data = 0; + c->turn_data_exceeded = 0; + c->turn_usage_exceeded = 0; + status = compact_projection(c, 0, 1, TURN_TOOLS_ENABLED, NULL, error, + sizeof(error)); + if (status == TURN_IO_INTERRUPTED) + return (3); + if (status != TURN_IO_OK) { + emit_error(c, "fugu: compaction failed: %s", + error[0] != '\0' ? error : "unknown failure"); + return (1); + } return (0); } @@ -3962,12 +4545,12 @@ effective_window(struct coord *c) static void cmd_context(struct coord *c) { - const char *sys = c->conf->system != NULL ? c->conf->system : - fugu_system_prompt; + const struct buf *system = lead_system(c); struct turn_txn_projection projection; struct msg *m; - size_t sysb = strlen(sys), prose = 0, tools = 0, largest = 0; - size_t ctxb = c->ctx_personal.len + c->ctx_project.len; + size_t sysb = system->len, prose = 0, tools = 0, largest = 0; + size_t ctxb = c->conf->project_context ? + c->epoch.personal.len + c->epoch.project.len : 0; size_t compb = c->compaction.len; int64_t win, total; int nmsg; @@ -3993,8 +4576,7 @@ cmd_context(struct coord *c) if (mb > largest) largest = mb; } - total = est_tokens(sysb) + est_tokens(ctxb) + est_tokens(compb) + - est_tokens(prose) + est_tokens(tools); + total = est_tokens(sysb) + est_tokens(prose) + est_tokens(tools); win = effective_window(c); { @@ -4002,7 +4584,7 @@ cmd_context(struct coord *c) buf_init(&r); buf_addf(&r, "context estimate (byte-derived; not exact):\n"); - buf_addf(&r, " system prompt ~%lld\n", + buf_addf(&r, " system baseline ~%lld\n", (long long)est_tokens(sysb)); buf_addf(&r, " project context ~%lld\n", (long long)est_tokens(ctxb)); @@ -4052,7 +4634,9 @@ line_command(struct coord *c, const char *line, int *t if (strcmp(line, "/quit") == 0 || strcmp(line, "/exit") == 0) return (1); if (strcmp(line, "/clear") == 0) { - cmd_clear(c); + rc = cmd_clear(c); + if (turn_status != NULL && rc != 0) + *turn_status = rc == 3 ? -3 : -2; return (0); } if (strcmp(line, "/context") == 0) { @@ -4408,6 +4992,7 @@ coord_run(struct fugu_conf *cf, const char *prompt, si { static struct coord c; struct turn_txn_jobs turn_jobs; + struct journal_state restored; struct msglist seed; const char *libexec; char sessdir[1024], id[SESSION_ID_MAX]; @@ -4439,12 +5024,16 @@ coord_run(struct fugu_conf *cf, const char *prompt, si buf_init(&c.input); buf_init(&c.ctx_personal); buf_init(&c.ctx_project); + system_epoch_init(&c.epoch); + system_epoch_init(&c.epoch_rollback); + c.files = compaction_files_new(); buf_init(&c.compaction); - buf_init(&c.sysfull); + buf_init(&c.compaction_prose); buf_init(&c.ctx_data); buf_init(&c.steer); buf_init(&c.steer_pend); TAILQ_INIT(&seed); + journal_state_init(&restored); libexec = libexec_dir(); c.libexec = libexec; do { @@ -4480,8 +5069,9 @@ coord_run(struct fugu_conf *cf, const char *prompt, si if (journal_dir(sessdir, sizeof(sessdir)) == -1) fatalx("cannot use ~/.fugu/sessions"); if (resume_id != NULL) { - if (journal_replay(sessdir, resume_id, &seed) == -1) + if (journal_restore(sessdir, resume_id, &restored) == -1) fatalx("unknown session: %s", resume_id); + TAILQ_CONCAT(&seed, &restored.messages, entry); if ((c.journal = journal_open(sessdir, resume_id)) == NULL) fatalx("cannot reopen session %s", resume_id); } else { @@ -4503,7 +5093,8 @@ coord_run(struct fugu_conf *cf, const char *prompt, si else c.skills = skills_load(""); /* empty set */ } - context_pre_pledge(&c, resume_id != NULL, sessdir, resume_id); + context_pre_pledge(&c, resume_id != NULL, &restored); + journal_state_free(&restored); /* the drawer's saved-session snapshot, while rpath still allows * the enumeration (behavior.md 2.1) */ @@ -4527,7 +5118,7 @@ coord_run(struct fugu_conf *cf, const char *prompt, si event_dispatch(); /* until every worker is READY and PONGed */ /* the in-tree context file is read via fugu-tool (no rpath here) */ - context_post_bringup(&c, resume_id != NULL); + context_post_bringup(&c, resume_id != NULL, &seed); memset(&turn_jobs, 0, sizeof(turn_jobs)); turn_jobs.arg = &c; turn_jobs.begin = turn_jobs_begin; @@ -4612,8 +5203,12 @@ coord_run(struct fugu_conf *cf, const char *prompt, si buf_free(&c.input); buf_free(&c.ctx_personal); buf_free(&c.ctx_project); + system_epoch_free(&c.epoch); + system_epoch_free(&c.epoch_rollback); + compaction_files_free(c.files); + compaction_files_free(c.files_rollback); buf_free(&c.compaction); - buf_free(&c.sysfull); + buf_free(&c.compaction_prose); buf_free(&c.ctx_data); buf_free(&c.steer); buf_free(&c.steer_pend);