commit - f3f30b82b058e173347141ecf1ccdebb3a6b2a16
commit + bc1fdebe1d123a6a90bcb6e3d48f644a3f5c85c5
blob - 270f362460e38a84a8a0639ff31e28dcc042408a
blob + 2b3e8a927cc285b38200b0d8cee2a4588e33e1f1
--- src/fugu/Makefile
+++ src/fugu/Makefile
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
#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"
/* 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;
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)
(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,
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);
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);
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 ||
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);
}
}
-/*
- * 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);
* 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));
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));
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));
req.cache = 1;
req.cache_tail_msg = cache_tail_msg;
anthropic_build_request(body, &req);
+ return (0);
}
+ return (-1);
}
static void
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);
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 "
"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);
}
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;
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));
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) {
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);
}
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;
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);
{
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));
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) {
{
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];
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 {
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 {
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) */
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;
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);
