commit 36f2bf37d293ba47b0c2c529451cfae948bb8cf0 from: Isaac Meerleo date: Wed Jul 15 16:37:25 2026 UTC integrate lead turn transaction ownership commit - c5f3a1cbb149227b342688a63cb3375c4be6c547 commit + 36f2bf37d293ba47b0c2c529451cfae948bb8cf0 blob - 1a4b63676dc080f78a50128e54b7bce0dca9a89e blob + c878964fdd6faef25d24fdb12904056668e8e318 --- regress/turn/run.sh +++ regress/turn/run.sh @@ -609,6 +609,25 @@ ok "compact+clear: the summary is dropped after /clear $([ $? -ne 0 ]; echo $?) stop_stub +# A failed Compaction leaves both the live Projection and its acknowledged +# cursor unchanged. The synthetic instruction is Provider-only: it is not a +# pending message, and failure must neither record a compact event nor make the +# following Turn claim a different cache boundary. +rm -rf "$HOME/.fugu" +rm -f "$dir/reqout" +start_stub "OK ERROR OK" +write_conf "$stubport" +printf 'first question\n/compact\nsecond question\n/quit\n' > "$dir/lin" +run_fugu_in "$dir/lin" +ok "failed compact: line session continues" \ + $([ "$rc" -eq 0 ]; echo $?) +grep -q '"t":"compact"' "$sessions"/*.ndjson 2>/dev/null +ok "failed compact: no compact event was recorded" \ + $([ $? -ne 0 ]; echo $?) +"$obj/reqcheck" "$dir/reqout" +ok "failed compact: Projection and cursor survive unchanged" $? +stop_stub + # --- 26: /context accounts for the compaction summary (behavior.md 3) rm -rf "$HOME/.fugu" start_stub "OK SUMMARY" @@ -839,6 +858,32 @@ if [ -x "$obj/ptydrive" ]; then ok "picker: selected provider metadata drove the context window" $? stop_stub + # Request a listing and immediately submit a slow-tool prompt; delayed + # selection must land between its two Generations. The selection cools + # the Projection cursor but must not terminate the Turn transaction. + rm -rf "$HOME/.fugu" + rm -f "$dir/reqout" + start_stub "SLEEPTOOL OK" + write_conf "$stubport" + printf '/model\rhello\r\000\000model-b\r\000\000\000\000/quit\r' | \ + TERM=vt100 HOME=$HOME \ + FUGU_CONF=$dir/fugu.conf FUGU_LIBEXEC=$dir/libexec/fugu \ + FUGU_CA_FILE=$dir/cert.pem FUGU_RETRY_BASE_MS=50 \ + "$obj/ptydrive" "$fugu" >"$dir/pickerout" 2>&1 + rc=$? + ok "late picker: active Turn survives model selection" \ + $([ "$rc" -eq 0 ]; echo $?) + sed -n '/^--- SLEEPTOOL ---$/,/^--- OK ---$/p' "$dir/reqlog" | \ + grep -q '"model":"claude-sonnet-4-6"' + ok "late picker: first Generation started on the old model" $? + sed -n '/^--- OK ---$/,$p' "$dir/reqlog" | \ + grep -q '"model":"stub-model-b"' + ok "late picker: selection drives the next Generation" $? + n=$(grep -h '"outcome":"ok"' "$sessions"/*.ndjson 2>/dev/null | wc -l) + ok "late picker: exactly one active Turn commits" \ + $([ "$n" -eq 1 ]; echo $?) + stop_stub + # An explicit context_limit remains the first-priority denominator. rm -rf "$HOME/.fugu" start_stub "OK" blob - 7f06ece20e98b13ffd03d0d95533dac60072f8fc blob + d3a5b25b1e3d7c524fd577fc257e6efab6dba794 --- regress/turn_txn/turn_txn_test.c +++ regress/turn_txn/turn_txn_test.c @@ -404,6 +404,44 @@ test_sent_snapshots_advance_and_stale_tokens_fail(void } static void +test_active_cache_reset_cools_cursor_without_ending_turn(void) +{ + struct fixture f; + struct msglist seed; + struct turn_txn_projection before_reset, current; + + TAILQ_INIT(&seed); + fixture_new(&f, &seed); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "one")); + projection(&f, ¤t); + turn_txn_projection_sent(f.txn, ¤t); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "two")); + projection(&f, &before_reset); + CHECK(before_reset.message_count == 2 && + before_reset.cache_tail_msg == 0); + + turn_txn_cache_reset(f.txn); + projection(&f, ¤t); + CHECK(current.message_count == 2 && current.cache_tail_msg == -1 && + current.token != before_reset.token); + CHECK(fatal_projection_sent(f.txn, &before_reset)); + turn_txn_projection_sent(f.txn, ¤t); + turn_txn_accept(f.txn, text_message(ROLE_USER, "tool result")); + turn_txn_commit(f.txn, TURN_TXN_OK); + projection(&f, ¤t); + CHECK(current.message_count == 3 && current.cache_tail_msg == 1); + + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "drop")); + turn_txn_cache_reset(f.txn); + turn_txn_abandon(f.txn); + projection(&f, ¤t); + CHECK(current.message_count == 3 && current.cache_tail_msg == -1); + turn_txn_free(f.txn); +} + +static void test_foreign_snapshot_is_rejected(void) { struct fixture one, two; @@ -531,7 +569,7 @@ enum fatal_case { FATAL_COMMIT_IDLE, FATAL_COMMIT_OUTCOME, FATAL_ABANDON_IDLE, - FATAL_RESET_ACTIVE, + FATAL_RESET_NULL, FATAL_CLEAR_ACTIVE, FATAL_CLEAR_CONTEXT, FATAL_COMPACT_ACTIVE, @@ -601,9 +639,8 @@ run_fatal_case(enum fatal_case which) case FATAL_ABANDON_IDLE: turn_txn_abandon(f.txn); break; - case FATAL_RESET_ACTIVE: - turn_txn_begin(f.txn); - turn_txn_cache_reset(f.txn); + case FATAL_RESET_NULL: + turn_txn_cache_reset(NULL); break; case FATAL_CLEAR_ACTIVE: turn_txn_begin(f.txn); @@ -668,6 +705,7 @@ main(void) test_success_is_durable_before_job_commit(); test_abandon_aborts_jobs_then_rolls_back_pending(); test_sent_snapshots_advance_and_stale_tokens_fail(); + test_active_cache_reset_cools_cursor_without_ending_turn(); test_foreign_snapshot_is_rejected(); test_cap_does_not_claim_unsent_messages(); test_resume_clear_compact_and_cache_reset_ownership(); blob - 942a23cf5d400d1e79a25735bb7088635fd8235f blob + ed92b0e4be6d6b4f5aa68f39e28f6d85e962b64b --- src/fugu/coord.c +++ src/fugu/coord.c @@ -52,6 +52,8 @@ #include "skills.h" #include "tooldefs.h" #include "agentcfg.h" +#include "turn_txn.h" +#include "turn_txn_private.h" #define TURN_ITER_MAX 50 /* behavior.md section 5 */ #define MAX_TOOL_CALLS 64 /* tool_use blocks per assistant msg */ @@ -148,11 +150,9 @@ struct coord { int have_sigint_ev; int cancelled; - struct msglist conv; /* the canonical conversation */ + struct turn_txn *turn; /* Lead Projection transaction */ struct journal *journal; struct skillset *skills; - int turn_n; - int prev_msgs; /* count in the last request */ /* curses front end (behavior.md 2.1): output goes to fugu-tty */ int ui; /* terminal mode: render via tty */ @@ -372,6 +372,41 @@ tool_turn_ctl(struct coord *c, uint32_t type) } static void +turn_jobs_begin(void *arg) +{ + struct coord *c = arg; + + tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_BEGIN); +} + +static void +turn_jobs_stop_executor(void *arg) +{ + struct coord *c = arg; + + if (!c->tool_cancel_needed) + return; + (void)kill(c->kids[ROLE_TOOL].pid, SIGTERM); + c->tool_cancel_needed = 0; +} + +static void +turn_jobs_commit(void *arg) +{ + struct coord *c = arg; + + tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_COMMIT); +} + +static void +turn_jobs_abort(void *arg) +{ + struct coord *c = arg; + + tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_ABORT); +} + +static void cancel_signal(int sig, short event, void *arg) { struct coord *c = arg; @@ -417,14 +452,13 @@ post_stream_test_delay(void) * pending set, so a pre-commit Ctrl-C always abandons and a post-boundary * Ctrl-C cannot create a half-committed outcome. */ static int -commit_turn(struct coord *c, enum turn_outcome outcome) +commit_turn(struct coord *c, enum turn_txn_commit outcome) { sigset_t block, oldmask, pending; int interrupted = 0; if (!c->have_sigint_ev) { - journal_turn_end(c->journal, c->turn_n, outcome); - tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_COMMIT); + turn_txn_commit(c->turn, outcome); return (0); } (void)event_loop(EVLOOP_NONBLOCK); @@ -438,10 +472,8 @@ commit_turn(struct coord *c, enum turn_outcome outcome if (c->cancelled || sigismember(&pending, SIGINT)) { c->cancelled = 1; interrupted = 1; - } else { - journal_turn_end(c->journal, c->turn_n, outcome); - tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_COMMIT); - } + } else + turn_txn_commit(c->turn, outcome); if (sigprocmask(SIG_SETMASK, &oldmask, NULL) == -1) fatal("sigprocmask restore SIGINT"); return (interrupted ? -1 : 0); @@ -458,17 +490,6 @@ libexec_dir(void) } static int -msg_count(const struct msglist *l) -{ - struct msg *m; - int n = 0; - - TAILQ_FOREACH(m, l, entry) - n++; - return (n); -} - -static int protect_has(const struct buf *b, const char *path) { size_t off, len; @@ -1621,7 +1642,7 @@ ui_event(struct coord *c, struct imsg *imsg) c->provider_window = sel.context_window; free(c->conf->model); c->conf->model = xstrdup(id); - c->prev_msgs = 0; /* the cache starts cold (8) */ + turn_txn_cache_reset(c->turn); /* the cache starts cold (8) */ send_ui_config(c); /* the status bar shows it */ emit_note(c, "model: %s", id); break; @@ -2554,9 +2575,10 @@ context_post_bringup(struct coord *c, int resumed) rebuild_system(c); } -/* Translate the canonical conversation into the active wire dialect. */ +/* Translate a borrowed Projection into the active wire dialect. */ static void build_provider_request(struct coord *c, struct buf *body, + 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) { @@ -2569,7 +2591,8 @@ build_provider_request(struct coord *c, struct buf *bo req.systemlen = c->sysfull.len; req.tools = tools; req.ntools = ntools; - req.conv = &c->conv; + req.conv = messages; + req.suffix = suffix; openai_build_request(body, &req); } else { struct anthropic_req req; @@ -2581,7 +2604,8 @@ build_provider_request(struct coord *c, struct buf *bo req.systemlen = c->sysfull.len; req.tools = tools; req.ntools = ntools; - req.conv = &c->conv; + req.conv = messages; + req.suffix = suffix; req.cache = 1; req.cache_tail_msg = cache_tail_msg; anthropic_build_request(body, &req); @@ -2597,6 +2621,7 @@ static int send_turn_request(struct coord *c) { const struct tool_def *base, *agent; + struct turn_txn_projection projection; struct tool_def tools[20]; struct buf body; int nbase, ntools; @@ -2614,8 +2639,9 @@ send_turn_request(struct coord *c) tools[ntools++] = *agent; } buf_init(&body); - build_provider_request(c, &body, tools, ntools, - c->prev_msgs > 0 ? c->prev_msgs - 1 : -1); + turn_txn_projection(c->turn, &projection); + build_provider_request(c, &body, projection.messages, NULL, tools, + ntools, projection.cache_tail_msg); if (body.len > FUGU_REQUEST_MAX) { /* * JSON escaping can expand piped context or a large tool @@ -2632,8 +2658,8 @@ send_turn_request(struct coord *c) return (-1); } send_request(c, &body, c->lead_api_id, c->active_provider_slot); + turn_txn_projection_sent(c->turn, &projection); buf_free(&body); - c->prev_msgs = msg_count(&c->conv); /* messages just sent */ return (0); } @@ -3163,56 +3189,33 @@ run_pending_tools(struct coord *c) return (um); } -/* - * Discard every canonical message appended by an unsuccessful turn. - * The Journal already treats TURN_ABANDONED transactionally; the live - * Projection must have the identical last-successful state (I14). - */ static int -abandon_turn(struct coord *c, struct msg *first, int status) +abandon_turn(struct coord *c, int status) { - struct msg *m, *next; - - /* A signal or drawer switch can break the Coordinator's wait while - * fugu-tool is still inside a synchronous executor. Its SIGTERM hook - * kills the published child group and lets dispatch resume to consume - * the ordered ABORT. */ - if (c->tool_cancel_needed) { - (void)kill(c->kids[ROLE_TOOL].pid, SIGTERM); - c->tool_cancel_needed = 0; - } - tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_ABORT); - journal_turn_end(c->journal, c->turn_n, TURN_ABANDONED); - for (m = first; m != NULL; m = next) { - next = TAILQ_NEXT(m, entry); - TAILQ_REMOVE(&c->conv, m, entry); - msg_free(m); - } + turn_txn_abandon(c->turn); reset_tcalls(c); - c->prev_msgs = 0; /* failed requests restart caching cold */ buf_reset(&c->steer); buf_reset(&c->steer_pend); return (status); } static int -cancel_turn(struct coord *c, struct msg *first) +cancel_turn(struct coord *c) { int status; - status = abandon_turn(c, first, 3); + status = abandon_turn(c, 3); machine_error(c, "turn cancelled"); machine_finish(c, 3); return (status); } static int -fail_turn(struct coord *c, struct msg *first, int status, - const char *message) +fail_turn(struct coord *c, int status, const char *message) { int result; - result = abandon_turn(c, first, status); + result = abandon_turn(c, status); machine_error(c, message); machine_finish(c, status); return (result); @@ -3230,7 +3233,7 @@ fail_turn(struct coord *c, struct msg *first, int stat static int run_turn(struct coord *c, const char *prompt, size_t promptlen) { - struct msg *um, *am, *first; + struct msg *um, *am; int iter; if (promptlen > FUGU_TURN_DATA_MAX) { @@ -3240,22 +3243,18 @@ run_turn(struct coord *c, const char *prompt, size_t p return (1); } um = msg_new(ROLE_USER); - first = um; msg_add_text(um, prompt, promptlen); - TAILQ_INSERT_TAIL(&c->conv, um, entry); - c->turn_n++; memset(&c->turn_usage, 0, sizeof(c->turn_usage)); c->turn_tool_calls = 0; c->turn_data = promptlen; c->turn_data_exceeded = 0; - journal_turn_begin(c->journal, c->turn_n); - tool_turn_ctl(c, FUGU_IMSG_TOOL_TURN_BEGIN); - journal_message(c->journal, um); + turn_txn_begin(c->turn); + turn_txn_accept(c->turn, um); if (cancel_checkpoint(c)) - return (cancel_turn(c, first)); + return (cancel_turn(c)); if (c->want_switch) { - int status = abandon_turn(c, first, 1); + int status = abandon_turn(c, 1); machine_error(c, "turn abandoned by session switch"); machine_finish(c, 1); @@ -3264,16 +3263,16 @@ run_turn(struct coord *c, const char *prompt, size_t p for (iter = 1; ; iter++) { if (send_turn_request(c) == -1) { - return (fail_turn(c, first, 1, + return (fail_turn(c, 1, "request serialization failed")); } pump_stream(c); post_stream_test_delay(); if (cancel_checkpoint(c)) - return (cancel_turn(c, first)); + return (cancel_turn(c)); if (c->turn_data_exceeded) { - int status = abandon_turn(c, first, 1); + int status = abandon_turn(c, 1); emit_error(c, "fugu: turn data exceeds 4 MiB"); machine_error(c, "turn data exceeds 4 MiB"); @@ -3283,13 +3282,13 @@ run_turn(struct coord *c, const char *prompt, size_t p /* a drawer switch abandons the in-flight turn cleanly * (behavior.md 2.1, section 7); its steer text dies with it */ if (c->want_switch) { - return (fail_turn(c, first, 1, + return (fail_turn(c, 1, "turn abandoned by session switch")); } if (c->failed) { const char *why = c->err[0] != '\0' ? c->err : "turn failed"; - int status = abandon_turn(c, first, 1); + int status = abandon_turn(c, 1); /* a failed turn contributes nothing (I14) */ emit_error(c, "fugu: %s", why); @@ -3300,7 +3299,7 @@ run_turn(struct coord *c, const char *prompt, size_t p /* neither text nor a tool call is empty content the * provider would reject on the next request (I14) */ if (c->text.len == 0 && c->ntcalls == 0) { - int status = abandon_turn(c, first, 1); + int status = abandon_turn(c, 1); emit_error(c, "fugu: provider sent an empty reply"); machine_error(c, "provider sent an empty reply"); @@ -3330,7 +3329,7 @@ run_turn(struct coord *c, const char *prompt, size_t p } } if (bad) { - int status = abandon_turn(c, first, 1); + int status = abandon_turn(c, 1); emit_error(c, "fugu: provider sent malformed tool arguments"); @@ -3341,14 +3340,13 @@ run_turn(struct coord *c, const char *prompt, size_t p } } if (cancel_checkpoint(c)) - return (cancel_turn(c, first)); + return (cancel_turn(c)); am = build_assistant_msg(c); - TAILQ_INSERT_TAIL(&c->conv, am, entry); - journal_message(c->journal, am); + turn_txn_accept(c->turn, am); machine_assistant(c); if (cancel_checkpoint(c)) - return (cancel_turn(c, first)); + return (cancel_turn(c)); /* * Any tool_use blocks MUST be answered with tool_result on @@ -3359,8 +3357,8 @@ run_turn(struct coord *c, const char *prompt, size_t p */ if (c->ntcalls == 0) { /* the model is done: commit and show its text */ - if (commit_turn(c, TURN_OK) == -1) - return (cancel_turn(c, first)); + if (commit_turn(c, TURN_TXN_OK) == -1) + return (cancel_turn(c)); emit_answer(c); machine_finish(c, 0); return (0); @@ -3392,17 +3390,16 @@ run_turn(struct coord *c, const char *prompt, size_t p c->turn_data_exceeded ? "turn data or worker result exceeded its bound" : "turn abandoned by session switch"; - return (fail_turn(c, first, status, why)); + return (fail_turn(c, status, why)); } - TAILQ_INSERT_TAIL(&c->conv, um, entry); - journal_message(c->journal, um); + turn_txn_accept(c->turn, um); reset_tcalls(c); if (cancel_checkpoint(c)) - return (cancel_turn(c, first)); + return (cancel_turn(c)); if (iter >= TURN_ITER_MAX) { - if (commit_turn(c, TURN_CAP) == -1) - return (cancel_turn(c, first)); + if (commit_turn(c, TURN_TXN_CAP) == -1) + return (cancel_turn(c)); emit_error(c, "fugu: reached the %d-step tool limit " "for one turn", TURN_ITER_MAX); machine_error(c, "tool-loop iteration limit reached"); @@ -3418,11 +3415,8 @@ static int64_t est_tokens(size_t); /* defined below */ static void cmd_clear(struct coord *c) { - journal_clear(c->journal); /* resume also starts fresh */ - msglist_free(&c->conv); - TAILQ_INIT(&c->conv); - c->prev_msgs = 0; /* cold cache after a reset */ - buf_reset(&c->compaction); /* a reset voids the summary */ + struct turn_txn_context context; + struct turn_txn_context *contextp = NULL; /* the in-tree file is re-read after a clear (behavior.md 4) */ if (c->conf->project_context) { @@ -3432,11 +3426,14 @@ 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"); - journal_context(c->journal, - (const char *)c->ctx_personal.data, c->ctx_personal.len, - (const char *)c->ctx_project.data, c->ctx_project.len); + 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; } - journal_sync(c->journal); + turn_txn_clear(c->turn, contextp); + buf_reset(&c->compaction); /* a reset voids the summary */ rebuild_system(c); /* drop the summary either way */ if (c->ui) ui_send(c, FUGU_IMSG_UI_CLEARED, NULL, 0); @@ -3452,27 +3449,28 @@ cmd_clear(struct coord *c) static int cmd_compact(struct coord *c) { + struct turn_txn_projection projection; struct msg *instr; struct buf body; - if (TAILQ_EMPTY(&c->conv)) { + 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) */ - /* a synthetic user turn carrying the summarization instruction */ + /* A Provider-only suffix carrying the summarization instruction. */ instr = msg_new(ROLE_USER); msg_add_text(instr, COMPACT_INSTR, strlen(COMPACT_INSTR)); - TAILQ_INSERT_TAIL(&c->conv, instr, entry); buf_init(&body); - build_provider_request(c, &body, NULL, 0, -1); /* no tools (S3) */ + build_provider_request(c, &body, projection.messages, instr, NULL, 0, + -1); /* no tools (S3) */ + msg_free(instr); if (body.len > FUGU_REQUEST_MAX) { buf_free(&body); - TAILQ_REMOVE(&c->conv, instr, entry); - msg_free(instr); emit_error(c, "fugu: too large to compact"); return (1); } @@ -3481,8 +3479,6 @@ cmd_compact(struct coord *c) c->turn_data = 0; c->turn_data_exceeded = 0; pump_stream(c); - TAILQ_REMOVE(&c->conv, instr, entry); - msg_free(instr); if (cancel_checkpoint(c)) return (3); @@ -3494,15 +3490,10 @@ cmd_compact(struct coord *c) if (cancel_checkpoint(c)) return (3); - /* the summary replaces the conversation and becomes context */ + /* 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); - msglist_free(&c->conv); - TAILQ_INIT(&c->conv); - c->prev_msgs = 0; - journal_compact(c->journal, (const char *)c->compaction.data, - c->compaction.len); - journal_sync(c->journal); rebuild_system(c); if (c->ui) ui_send(c, FUGU_IMSG_UI_CLEARED, NULL, 0); @@ -3588,14 +3579,17 @@ cmd_context(struct coord *c) { const char *sys = c->conf->system != NULL ? c->conf->system : fugu_system_prompt; + 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 compb = c->compaction.len; int64_t win, total; - int nmsg = 0; + int nmsg; - TAILQ_FOREACH(m, &c->conv, entry) { + turn_txn_projection(c->turn, &projection); + nmsg = (int)projection.message_count; + TAILQ_FOREACH(m, projection.messages, entry) { struct block *b; size_t mb = 0; @@ -3613,7 +3607,6 @@ cmd_context(struct coord *c) } if (mb > largest) largest = mb; - nmsg++; } total = est_tokens(sysb) + est_tokens(ctxb) + est_tokens(compb) + est_tokens(prose) + est_tokens(tools); @@ -4022,6 +4015,8 @@ coord_run(struct fugu_conf *cf, const char *prompt, si enum output_mode output_mode, int debug, int verbose) { static struct coord c; + struct turn_txn_jobs turn_jobs; + struct msglist seed; const char *libexec; char sessdir[1024], id[SESSION_ID_MAX]; unsigned int i; @@ -4057,7 +4052,7 @@ coord_run(struct fugu_conf *cf, const char *prompt, si buf_init(&c.ctx_data); buf_init(&c.steer); buf_init(&c.steer_pend); - TAILQ_INIT(&c.conv); + TAILQ_INIT(&seed); libexec = libexec_dir(); c.libexec = libexec; do { @@ -4088,13 +4083,12 @@ coord_run(struct fugu_conf *cf, const char *prompt, si c.nkids++; } - /* Open the journal now (still unrestricted). Resume replays - * prior turns; the cache starts cold on resume (behavior.md 8), - * so prev_msgs stays 0. */ + /* Open the Journal now (still unrestricted). Resume replays into the + * seed; the Turn transaction constructor starts its cursor cold. */ if (journal_dir(sessdir, sizeof(sessdir)) == -1) fatalx("cannot use ~/.fugu/sessions"); if (resume_id != NULL) { - if (journal_replay(sessdir, resume_id, &c.conv) == -1) + if (journal_replay(sessdir, resume_id, &seed) == -1) fatalx("unknown session: %s", resume_id); if ((c.journal = journal_open(sessdir, resume_id)) == NULL) fatalx("cannot reopen session %s", resume_id); @@ -4142,6 +4136,13 @@ coord_run(struct fugu_conf *cf, const char *prompt, si /* the in-tree context file is read via fugu-tool (no rpath here) */ context_post_bringup(&c, resume_id != NULL); + memset(&turn_jobs, 0, sizeof(turn_jobs)); + turn_jobs.arg = &c; + turn_jobs.begin = turn_jobs_begin; + turn_jobs.stop_executor_if_live = turn_jobs_stop_executor; + turn_jobs.commit = turn_jobs_commit; + turn_jobs.abort = turn_jobs_abort; + c.turn = turn_txn_new(c.journal, &seed, &turn_jobs); if (prompt != NULL) { signal_set(&c.sigint_ev, SIGINT, cancel_signal, &c); @@ -4161,6 +4162,8 @@ coord_run(struct fugu_conf *cf, const char *prompt, si if (c.have_sigint_ev) signal_del(&c.sigint_ev); coord_teardown(&c); + turn_txn_free(c.turn); + c.turn = NULL; journal_close(c.journal); close(c.devnull_fd); c.devnull_fd = -1; @@ -4198,7 +4201,6 @@ coord_run(struct fugu_conf *cf, const char *prompt, si skills_free(c.skills); agentcfg_free(c.agentcfg); - msglist_free(&c.conv); reset_tcalls(&c); free(c.sessions); free(c.listed_models); blob - cf405af16f58485c538794076e0a4edde1b491cb blob + 78d9dfe99abc788cbd205f9c9dd5c32be8b083de --- src/fugu/turn_txn.c +++ src/fugu/turn_txn.c @@ -241,7 +241,7 @@ turn_txn_abandon(struct turn_txn *txn) void turn_txn_cache_reset(struct turn_txn *txn) { - require_idle(txn); + require_txn(txn); txn->sent_prefix_len = 0; snapshot_changed(txn); } blob - 0d511e884776f1230014b121031ad3012c6b5294 blob + bb2d9bf523aa772d6df503f3c7dfa1ba0ae55fc7 --- src/fugu/turn_txn.h +++ src/fugu/turn_txn.h @@ -59,6 +59,10 @@ void turn_txn_projection_sent(struct turn_txn *, const struct turn_txn_projection *); void turn_txn_commit(struct turn_txn *, enum turn_txn_commit); void turn_txn_abandon(struct turn_txn *); +/* + * Make the Projection cursor cold without changing transaction state. + * Valid while idle or active; invalidates every outstanding receipt. + */ void turn_txn_cache_reset(struct turn_txn *); void turn_txn_clear(struct turn_txn *, const struct turn_txn_context *); void turn_txn_compact(struct turn_txn *, const void *, size_t);