commit 6700e6d8bc89354c9145ee84ad4d8a3a666cc8c2 from: Isaac Meerleo date: Wed Jul 15 15:39:28 2026 UTC add lead turn transaction module commit - 91b632582edaf81718fbffc9193f46706b137bd2 commit + 6700e6d8bc89354c9145ee84ad4d8a3a666cc8c2 blob - 526917744516c4cc3b90e94fbe98a5f519ed063a blob + 8684989960ba557e858f510879a21bf60bdc7263 --- CONTEXT.md +++ CONTEXT.md @@ -27,6 +27,16 @@ _Avoid_: model turn, provider turn The conversation state as shaped for and sent to a provider's API. _Avoid_: history, context (alone) +**Pending Projection**: +The messages accepted during an active Lead Turn but not yet made durable; +later Generations in that Turn see them, while abandonment discards them all. +_Avoid_: temporary history, uncommitted conversation + +**Projection cursor**: +The last message boundary included in a completely sent Projection; a cold +cursor means the next Provider request claims no prior cache boundary. +_Avoid_: previous messages, cache index + **Compaction**: Replacing the active context with a model-written summary (`/compact`). _Avoid_: summarization (alone) @@ -52,6 +62,11 @@ replaying it reproduces exactly the state after the la turn (I14). _Avoid_: log, history file +**Turn transaction**: +The all-or-nothing persistence span of one Lead Turn: its Pending Projection +either survives together on success or disappears together on abandonment. +_Avoid_: journal transaction (when the whole Turn is meant) + **Freshness guard**: The changed-since-read refusal: mutating tools fstat the opened fd and refuse when size/mtime differ from the recorded read. blob - cb2cb4504dfaeeff1442b53c94c8e823f6989863 blob + 09b03732599422322cc16ad000349f3775b5efef --- regress/Makefile +++ regress/Makefile @@ -1,5 +1,5 @@ SUBDIR= agentcfg anthropic buf conf deploy diff editor generation http imsgev journal json log markdown \ model_window openai output print sandbox skills sse term tools turn ui_config \ - web xmalloc + turn_txn web xmalloc .include blob - 57c2435bf5f099f573b300ffbe187a7635b10368 blob + fe102d4dafbcf9bd82f474cd2053e352e8bcc847 --- regress/README +++ regress/README @@ -79,7 +79,9 @@ I13 length-aware strings buf, json (NUL round-trip), I14 resume fidelity journal (transactional replay: failed/abandoned/crash turns contribute nothing; strict paired - ids and listing parity); turn + ids and listing parity); turn_txn + (pending Projection, durable terminals, + and sent-prefix cursor); turn (end-to-end -c resume over TLS; an empty reply is abandoned, not committed as an unusable message); @@ -181,6 +183,8 @@ src/fugu/agentcfg.c agentcfg; turn (live schema/rout src/fugu/priv.c, coord.c sandbox (spawn/handshake/teardown), turn (lead and subagent turn loops) src/fugu/journal.c journal, turn +src/fugu/turn_txn.c turn_txn (pure ordering plus real-Journal + integration); turn (cache/SIGINT path) src/fugu-tty/ui_config_rx.c ui_config; turn (full-palette PTY path) src/common/imsgev.c imsgev (whole-payload views, checked normal/teardown sends, event failures); blob - /dev/null blob + 9cc696317e616a67a18200fbffa683dc1c4be6b2 (mode 644) --- /dev/null +++ regress/turn_txn/Makefile @@ -0,0 +1,9 @@ +PROGS= turn_txn_test turn_txn_journal_test +SRCS_turn_txn_test= turn_txn_test.c turn_txn.c msg.c log.c xmalloc.c +SRCS_turn_txn_journal_test=turn_txn_journal_test.c turn_txn.c journal.c \ + msg.c json.c buf.c log.c xmalloc.c + +.PATH: ${.CURDIR}/../../src/fugu +CFLAGS+= -I${.CURDIR}/../../src/fugu + +.include blob - /dev/null blob + 8cdfe1972919af2a12f240b2233d7d637a1bc04b (mode 644) --- /dev/null +++ regress/turn_txn/turn_txn_journal_test.c @@ -0,0 +1,255 @@ +/* + * Copyright (c) 2026 Isaac + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +/* Real Journal Adapter gate for ADR-0005 transaction ownership. */ + +#include +#include + +#include +#include +#include +#include +#include + +#include "buf.h" +#include "journal.h" +#include "log.h" +#include "msg.h" +#include "turn_txn.h" +#include "turn_txn_private.h" + +#include "regress.h" + +static char dir[] = "/tmp/turn_txn_journal.XXXXXXXXXX"; + +struct job_trace { + int begin; + int stop_checks; + int commit; + int abort; +}; + +static void +job_begin(void *arg) +{ + ((struct job_trace *)arg)->begin++; +} + +static void +job_stop_if_live(void *arg) +{ + ((struct job_trace *)arg)->stop_checks++; +} + +static void +job_commit(void *arg) +{ + ((struct job_trace *)arg)->commit++; +} + +static void +job_abort(void *arg) +{ + ((struct job_trace *)arg)->abort++; +} + +static void +jobs_init(struct turn_txn_jobs *jobs, struct job_trace *trace) +{ + memset(jobs, 0, sizeof(*jobs)); + jobs->arg = trace; + jobs->begin = job_begin; + jobs->stop_executor_if_live = job_stop_if_live; + jobs->commit = job_commit; + jobs->abort = job_abort; +} + +static struct msg * +text_message(enum msg_role role, const char *text) +{ + struct msg *message; + + message = msg_new(role); + msg_add_text(message, text, strlen(text)); + return (message); +} + +static size_t +message_count(const struct msglist *messages) +{ + struct msg *message; + size_t count = 0; + + TAILQ_FOREACH(message, messages, entry) + count++; + return (count); +} + +static const char * +message_text(const struct msglist *messages, size_t index) +{ + struct msg *message; + struct block *block; + size_t i = 0; + + TAILQ_FOREACH(message, messages, entry) { + if (i++ != index) + continue; + block = TAILQ_FIRST(&message->blocks); + return (block == NULL ? NULL : block->text); + } + return (NULL); +} + +static struct turn_txn * +session_new(struct journal **journal, char *id, struct job_trace *trace) +{ + struct turn_txn_jobs jobs; + struct msglist seed; + + TAILQ_INIT(&seed); + *journal = journal_create(dir, id, SESSION_ID_MAX); + if (*journal == NULL) + err(1, "journal_create"); + jobs_init(&jobs, trace); + return (turn_txn_new(*journal, &seed, &jobs)); +} + +static void +accept_pair(struct turn_txn *txn, const char *user, const char *assistant, + enum turn_txn_commit outcome) +{ + turn_txn_begin(txn); + turn_txn_accept(txn, text_message(ROLE_USER, user)); + turn_txn_accept(txn, text_message(ROLE_ASSISTANT, assistant)); + turn_txn_commit(txn, outcome); +} + +static void +test_real_journal_terminals(void) +{ + struct job_trace trace = {0}; + struct journal *journal; + struct turn_txn *txn; + struct msglist replay; + char id[SESSION_ID_MAX]; + + txn = session_new(&journal, id, &trace); + accept_pair(txn, "ok user", "ok answer", TURN_TXN_OK); + accept_pair(txn, "cap user", "cap answer", TURN_TXN_CAP); + turn_txn_begin(txn); + turn_txn_accept(txn, text_message(ROLE_USER, "abandoned user")); + turn_txn_accept(txn, text_message(ROLE_ASSISTANT, "abandoned answer")); + turn_txn_abandon(txn); + CHECK(trace.begin == 3 && trace.commit == 2 && trace.abort == 1 && + trace.stop_checks == 1); + turn_txn_free(txn); + journal_close(journal); + + TAILQ_INIT(&replay); + CHECK(journal_replay(dir, id, &replay) == 0); + CHECK(message_count(&replay) == 4); + CHECK(message_text(&replay, 0) != NULL && + strcmp(message_text(&replay, 0), "ok user") == 0); + CHECK(message_text(&replay, 1) != NULL && + strcmp(message_text(&replay, 1), "ok answer") == 0); + CHECK(message_text(&replay, 2) != NULL && + strcmp(message_text(&replay, 2), "cap user") == 0); + CHECK(message_text(&replay, 3) != NULL && + strcmp(message_text(&replay, 3), "cap answer") == 0); + msglist_free(&replay); +} + +static void +test_real_journal_resets(void) +{ + static const char personal[] = { 'p', '\0', 'x' }; + static const char project[] = { 'r', '\0', 'y' }; + static const char summary[] = { 's', '\0', 'm' }; + struct turn_txn_context context; + struct turn_txn_jobs jobs; + struct job_trace trace = {0}; + struct journal *journal; + struct turn_txn *txn; + struct msglist replay; + struct buf p, r, compact; + char id[SESSION_ID_MAX]; + + txn = session_new(&journal, id, &trace); + accept_pair(txn, "before clear", "old", TURN_TXN_OK); + context.personal = personal; + context.personal_len = sizeof(personal); + context.project = project; + context.project_len = sizeof(project); + turn_txn_clear(txn, &context); + turn_txn_free(txn); + journal_close(journal); + TAILQ_INIT(&replay); + CHECK(journal_replay(dir, id, &replay) == 0); + CHECK(TAILQ_EMPTY(&replay)); + buf_init(&p); + buf_init(&r); + buf_init(&compact); + CHECK(journal_replay_context(dir, id, &p, &r, &compact) == 0); + CHECK(p.len == sizeof(personal) && + memcmp(p.data, personal, sizeof(personal)) == 0); + CHECK(r.len == sizeof(project) && + memcmp(r.data, project, sizeof(project)) == 0); + CHECK(compact.len == 0); + buf_free(&p); + buf_free(&r); + buf_free(&compact); + + memset(&trace, 0, sizeof(trace)); + TAILQ_INIT(&replay); + journal = journal_create(dir, id, sizeof(id)); + if (journal == NULL) + err(1, "journal_create compact"); + jobs_init(&jobs, &trace); + txn = turn_txn_new(journal, &replay, &jobs); + accept_pair(txn, "before compact", "old", TURN_TXN_OK); + turn_txn_compact(txn, summary, sizeof(summary)); + turn_txn_free(txn); + journal_close(journal); + TAILQ_INIT(&replay); + CHECK(journal_replay(dir, id, &replay) == 0); + CHECK(TAILQ_EMPTY(&replay)); + buf_init(&p); + buf_init(&r); + buf_init(&compact); + CHECK(journal_replay_context(dir, id, &p, &r, &compact) == 0); + CHECK(compact.len == sizeof(summary) && + memcmp(compact.data, summary, sizeof(summary)) == 0); + buf_free(&p); + buf_free(&r); + buf_free(&compact); +} + +int +main(void) +{ + char cmd[1100]; + + log_init(LOG_TO_STDERR, 0, LOG_USER); + if (mkdtemp(dir) == NULL) + err(1, "mkdtemp"); + test_real_journal_terminals(); + test_real_journal_resets(); + (void)snprintf(cmd, sizeof(cmd), "rm -rf %s", dir); + (void)system(cmd); + REGRESS_END(); +} blob - /dev/null blob + 7f06ece20e98b13ffd03d0d95533dac60072f8fc (mode 644) --- /dev/null +++ regress/turn_txn/turn_txn_test.c @@ -0,0 +1,676 @@ +/* + * Copyright (c) 2026 Isaac + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +/* + * Pure Turn transaction suite (ADR-0005): Projection ownership, matching + * Journal and job transactions, durable terminal ordering, and the request + * cache cursor. Calibration moved job COMMIT before the Journal sync (the + * success-order check failed), then advanced the cursor at commit (the OK, + * CAP, and resumed-prefix checks failed); both mutations were restored. + */ + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "log.h" +#include "msg.h" +#include "journal.h" +#include "turn_txn.h" +#include "turn_txn_private.h" + +#include "regress.h" + +struct fixture { + struct turn_txn *txn; + char trace[1024]; + int turn_begin_n; + int turn_end_n; + enum turn_outcome turn_end_outcome; + int executor_live; + int message_calls; + size_t message_seen[16]; + int sync_calls; + size_t sync_seen; + u_char personal[32]; + size_t personal_len; + u_char project[32]; + size_t project_len; + u_char summary[32]; + size_t summary_len; +}; + +static void +trace_add(struct fixture *f, const char *event) +{ + size_t len; + + len = strlen(f->trace); + if (len != 0) + (void)snprintf(f->trace + len, sizeof(f->trace) - len, ","); + len = strlen(f->trace); + (void)snprintf(f->trace + len, sizeof(f->trace) - len, "%s", event); +} + +static void +projection(struct fixture *f, struct turn_txn_projection *p) +{ + memset(p, 0, sizeof(*p)); + turn_txn_projection(f->txn, p); +} + +void +journal_turn_begin(struct journal *journal, int n) +{ + struct fixture *f = (void *)journal; + + f->turn_begin_n = n; + trace_add(f, "journal.begin"); +} + +void +journal_message(struct journal *journal, const struct msg *m) +{ + struct fixture *f = (void *)journal; + struct turn_txn_projection p; + + (void)m; + projection(f, &p); + if (f->message_calls < (int)(sizeof(f->message_seen) / + sizeof(f->message_seen[0]))) + f->message_seen[f->message_calls] = p.message_count; + f->message_calls++; + trace_add(f, "journal.message"); +} + +void +journal_turn_end(struct journal *journal, int n, enum turn_outcome outcome) +{ + struct fixture *f = (void *)journal; + struct turn_txn_projection p; + + f->turn_end_n = n; + f->turn_end_outcome = outcome; + switch (outcome) { + case TURN_OK: + trace_add(f, "journal.end.ok"); + break; + case TURN_CAP: + trace_add(f, "journal.end.cap"); + break; + case TURN_ABANDONED: + trace_add(f, "journal.end.abandoned"); + break; + } + /* The production terminal callback includes journal fsync. */ + projection(f, &p); + f->sync_calls++; + f->sync_seen = p.message_count; + trace_add(f, "journal.sync"); +} + +void +journal_sync(struct journal *journal) +{ + struct fixture *f = (void *)journal; + struct turn_txn_projection p; + + projection(f, &p); + f->sync_calls++; + f->sync_seen = p.message_count; + trace_add(f, "journal.sync"); +} + +void +journal_clear(struct journal *journal) +{ + trace_add((void *)journal, "journal.clear"); +} + +void +journal_context(struct journal *journal, const char *personal, + size_t personal_len, + const char *project, size_t project_len) +{ + struct fixture *f = (void *)journal; + + f->personal_len = personal_len; + f->project_len = project_len; + if (personal_len <= sizeof(f->personal) && personal_len > 0) + memcpy(f->personal, personal, personal_len); + if (project_len <= sizeof(f->project) && project_len > 0) + memcpy(f->project, project, project_len); + trace_add(f, "journal.context"); +} + +void +journal_compact(struct journal *journal, const char *summary, + size_t summary_len) +{ + struct fixture *f = (void *)journal; + + f->summary_len = summary_len; + if (summary_len <= sizeof(f->summary) && summary_len > 0) + memcpy(f->summary, summary, summary_len); + trace_add(f, "journal.compact"); +} + +static void +jobs_begin(void *arg) +{ + trace_add(arg, "jobs.begin"); +} + +static void +jobs_stop_if_live(void *arg) +{ + struct fixture *f = arg; + + if (!f->executor_live) + return; + f->executor_live = 0; + trace_add(f, "jobs.stop"); +} + +static void +jobs_commit(void *arg) +{ + trace_add(arg, "jobs.commit"); +} + +static void +jobs_abort(void *arg) +{ + trace_add(arg, "jobs.abort"); +} + +static void +fixture_jobs(struct fixture *f, struct turn_txn_jobs *jobs) +{ + memset(jobs, 0, sizeof(*jobs)); + jobs->arg = f; + jobs->begin = jobs_begin; + jobs->stop_executor_if_live = jobs_stop_if_live; + jobs->commit = jobs_commit; + jobs->abort = jobs_abort; +} + +static void +fixture_new(struct fixture *f, struct msglist *seed) +{ + struct turn_txn_jobs jobs; + + memset(f, 0, sizeof(*f)); + fixture_jobs(f, &jobs); + f->txn = turn_txn_new((struct journal *)f, seed, &jobs); +} + +static struct msg * +text_message(enum msg_role role, const char *text) +{ + struct msg *m; + + m = msg_new(role); + msg_add_text(m, text, strlen(text)); + return (m); +} + +static void +append_text(struct msglist *messages, enum msg_role role, const char *text) +{ + struct msg *m; + + m = text_message(role, text); + TAILQ_INSERT_TAIL(messages, m, entry); +} + +static void +trace_reset(struct fixture *f) +{ + f->trace[0] = '\0'; +} + +static const char * +first_text(const struct msglist *messages) +{ + struct msg *m; + struct block *b; + + m = TAILQ_FIRST(messages); + if (m == NULL || (b = TAILQ_FIRST(&m->blocks)) == NULL || + b->type != BLOCK_TEXT) + return (NULL); + return (b->text); +} + +static int +fatal_projection_sent(struct turn_txn *txn, + const struct turn_txn_projection *projection) +{ + pid_t pid; + int status; + + fflush(NULL); + if ((pid = fork()) == -1) + return (0); + if (pid == 0) { + (void)close(STDERR_FILENO); + (void)open("/dev/null", O_WRONLY); + turn_txn_projection_sent(txn, projection); + _exit(0); + } + if (waitpid(pid, &status, 0) != pid) + return (0); + return (WIFEXITED(status) && WEXITSTATUS(status) == 1); +} + +static void +test_success_is_durable_before_job_commit(void) +{ + struct fixture f; + struct msglist seed; + struct msg *seed_msg; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + seed_msg = text_message(ROLE_USER, "resumed"); + TAILQ_INSERT_TAIL(&seed, seed_msg, entry); + fixture_new(&f, &seed); + CHECK(TAILQ_EMPTY(&seed)); + + projection(&f, &p); + CHECK(p.message_count == 1 && p.cache_tail_msg == -1 && + TAILQ_FIRST(p.messages) == seed_msg); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "question")); + projection(&f, &p); + CHECK(p.message_count == 2 && p.cache_tail_msg == -1); + turn_txn_projection_sent(f.txn, &p); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "answer")); + turn_txn_commit(f.txn, TURN_TXN_OK); + + CHECK(strcmp(f.trace, "journal.begin,jobs.begin,journal.message," + "journal.message,journal.end.ok,journal.sync,jobs.commit") == 0); + CHECK(f.turn_begin_n == 1 && f.turn_end_n == 1 && + f.turn_end_outcome == TURN_OK); + CHECK(f.message_calls == 2 && f.message_seen[0] == 1 && + f.message_seen[1] == 2); + CHECK(f.sync_calls == 1 && f.sync_seen == 3); + projection(&f, &p); + CHECK(p.message_count == 3 && p.cache_tail_msg == 1); + turn_txn_free(f.txn); +} + +static void +test_abandon_aborts_jobs_then_rolls_back_pending(void) +{ + struct fixture f; + struct msglist seed; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "keep"); + fixture_new(&f, &seed); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "drop question")); + projection(&f, &p); + turn_txn_projection_sent(f.txn, &p); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "drop answer")); + f.executor_live = 1; + turn_txn_abandon(f.txn); + + CHECK(strcmp(f.trace, "journal.begin,jobs.begin,journal.message," + "journal.message,jobs.stop,jobs.abort,journal.end.abandoned," + "journal.sync") == 0); + CHECK(f.sync_seen == 3); + projection(&f, &p); + CHECK(p.message_count == 1 && p.cache_tail_msg == -1); + CHECK(strcmp(first_text(p.messages), "keep") == 0); + + trace_reset(&f); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "next")); + turn_txn_commit(f.txn, TURN_TXN_OK); + CHECK(f.turn_begin_n == 2 && f.turn_end_n == 2); + CHECK(strstr(f.trace, "jobs.stop") == NULL); + trace_reset(&f); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "drop without executor")); + trace_reset(&f); + turn_txn_abandon(f.txn); + CHECK(strcmp(f.trace, "jobs.abort,journal.end.abandoned," + "journal.sync") == 0); + turn_txn_free(f.txn); +} + +static void +test_sent_snapshots_advance_and_stale_tokens_fail(void) +{ + struct fixture f; + struct msglist seed; + struct turn_txn_projection first, after_sent, 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, &first); + CHECK(first.message_count == 1 && first.cache_tail_msg == -1); + turn_txn_projection_sent(f.txn, &first); + projection(&f, &after_sent); + CHECK(after_sent.token != first.token && after_sent.cache_tail_msg == 0); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "two")); + CHECK(fatal_projection_sent(f.txn, &after_sent)); + projection(&f, ¤t); + CHECK(current.token != after_sent.token && current.message_count == 2 && + current.cache_tail_msg == 0); + current.cache_tail_msg = -1; + CHECK(fatal_projection_sent(f.txn, ¤t)); + projection(&f, ¤t); + current.message_count++; + CHECK(fatal_projection_sent(f.txn, ¤t)); + projection(&f, ¤t); + turn_txn_projection_sent(f.txn, ¤t); + turn_txn_commit(f.txn, TURN_TXN_OK); + projection(&f, ¤t); + CHECK(current.cache_tail_msg == 1); + turn_txn_cache_reset(f.txn); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "later Turn")); + CHECK(fatal_projection_sent(f.txn, ¤t)); + turn_txn_abandon(f.txn); + turn_txn_free(f.txn); +} + +static void +test_foreign_snapshot_is_rejected(void) +{ + struct fixture one, two; + struct msglist seed_one, seed_two; + struct turn_txn_projection foreign, current; + + TAILQ_INIT(&seed_one); + TAILQ_INIT(&seed_two); + fixture_new(&one, &seed_one); + fixture_new(&two, &seed_two); + turn_txn_begin(one.txn); + turn_txn_begin(two.txn); + turn_txn_accept(one.txn, text_message(ROLE_USER, "one")); + turn_txn_accept(two.txn, text_message(ROLE_USER, "two")); + projection(&one, &foreign); + projection(&two, ¤t); + CHECK(foreign.token == current.token); + CHECK(fatal_projection_sent(two.txn, &foreign)); + projection(&two, ¤t); + CHECK(current.cache_tail_msg == -1); + turn_txn_abandon(one.txn); + turn_txn_abandon(two.txn); + turn_txn_free(one.txn); + turn_txn_free(two.txn); +} + +static void +test_cap_does_not_claim_unsent_messages(void) +{ + struct fixture f; + struct msglist seed; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + fixture_new(&f, &seed); + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "request")); + projection(&f, &p); + turn_txn_projection_sent(f.txn, &p); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "tool call")); + turn_txn_accept(f.txn, text_message(ROLE_USER, "tool result")); + turn_txn_commit(f.txn, TURN_TXN_CAP); + CHECK(f.turn_end_outcome == TURN_CAP); + + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "next request")); + projection(&f, &p); + CHECK(p.message_count == 4 && p.cache_tail_msg == 0); + turn_txn_abandon(f.txn); + projection(&f, &p); + CHECK(p.message_count == 3 && p.cache_tail_msg == -1); + turn_txn_free(f.txn); +} + +static void +test_resume_clear_compact_and_cache_reset_ownership(void) +{ + static const char personal[] = { 'p', '\0', 'x' }; + static const char project[] = { 'r', '\0', 'y', 'z' }; + static const char summary[] = { 's', '\0', 'm' }; + struct turn_txn_context context; + struct fixture f; + struct msglist seed; + struct turn_txn_projection p; + + TAILQ_INIT(&seed); + append_text(&seed, ROLE_USER, "resume one"); + append_text(&seed, ROLE_ASSISTANT, "resume two"); + fixture_new(&f, &seed); + CHECK(TAILQ_EMPTY(&seed)); + projection(&f, &p); + CHECK(p.message_count == 2 && p.cache_tail_msg == -1 && + strcmp(first_text(p.messages), "resume one") == 0); + + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "sent")); + projection(&f, &p); + turn_txn_projection_sent(f.txn, &p); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "reply")); + turn_txn_commit(f.txn, TURN_TXN_OK); + projection(&f, &p); + CHECK(p.message_count == 4 && p.cache_tail_msg == 2); + turn_txn_cache_reset(f.txn); + projection(&f, &p); + CHECK(p.message_count == 4 && p.cache_tail_msg == -1); + + context.personal = personal; + context.personal_len = sizeof(personal); + context.project = project; + context.project_len = sizeof(project); + trace_reset(&f); + turn_txn_clear(f.txn, &context); + CHECK(strcmp(f.trace, "journal.clear,journal.context,journal.sync") == 0); + CHECK(f.sync_seen == 4); + CHECK(f.personal_len == sizeof(personal) && + memcmp(f.personal, personal, sizeof(personal)) == 0); + CHECK(f.project_len == sizeof(project) && + memcmp(f.project, project, sizeof(project)) == 0); + projection(&f, &p); + CHECK(p.message_count == 0 && p.cache_tail_msg == -1); + + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, text_message(ROLE_USER, "after clear")); + projection(&f, &p); + turn_txn_projection_sent(f.txn, &p); + turn_txn_accept(f.txn, text_message(ROLE_ASSISTANT, "after reply")); + turn_txn_commit(f.txn, TURN_TXN_OK); + trace_reset(&f); + turn_txn_compact(f.txn, summary, sizeof(summary)); + CHECK(strcmp(f.trace, "journal.compact,journal.sync") == 0); + CHECK(f.sync_seen == 2 && f.summary_len == sizeof(summary) && + memcmp(f.summary, summary, sizeof(summary)) == 0); + projection(&f, &p); + CHECK(p.message_count == 0 && p.cache_tail_msg == -1); + turn_txn_free(f.txn); +} + +enum fatal_case { + FATAL_BEGIN_ACTIVE, + FATAL_ACCEPT_IDLE, + FATAL_ACCEPT_NULL, + FATAL_PROJECTION_NULL, + FATAL_PROJECTION_SENT_IDLE, + FATAL_PROJECTION_SENT_NULL, + FATAL_COMMIT_IDLE, + FATAL_COMMIT_OUTCOME, + FATAL_ABANDON_IDLE, + FATAL_RESET_ACTIVE, + FATAL_CLEAR_ACTIVE, + FATAL_CLEAR_CONTEXT, + FATAL_COMPACT_ACTIVE, + FATAL_COMPACT_NULL, + FATAL_FREE_ACTIVE, + FATAL_NEW_NULL_SEED, + FATAL_NEW_NULL_JOURNAL, + FATAL_NEW_BAD_JOBS +}; + +static void +run_fatal_case(enum fatal_case which) +{ + struct turn_txn_jobs jobs; + struct turn_txn_context context; + struct turn_txn_projection p; + struct fixture f; + struct msglist seed; + + TAILQ_INIT(&seed); + memset(&f, 0, sizeof(f)); + fixture_jobs(&f, &jobs); + if (which == FATAL_NEW_NULL_SEED) { + (void)turn_txn_new((struct journal *)&f, NULL, &jobs); + return; + } + if (which == FATAL_NEW_NULL_JOURNAL) { + (void)turn_txn_new(NULL, &seed, &jobs); + return; + } + if (which == FATAL_NEW_BAD_JOBS) { + jobs.stop_executor_if_live = NULL; + (void)turn_txn_new((struct journal *)&f, &seed, &jobs); + return; + } + f.txn = turn_txn_new((struct journal *)&f, &seed, &jobs); + switch (which) { + case FATAL_BEGIN_ACTIVE: + turn_txn_begin(f.txn); + turn_txn_begin(f.txn); + break; + case FATAL_ACCEPT_IDLE: + turn_txn_accept(f.txn, text_message(ROLE_USER, "bad")); + break; + case FATAL_ACCEPT_NULL: + turn_txn_begin(f.txn); + turn_txn_accept(f.txn, NULL); + break; + case FATAL_PROJECTION_NULL: + turn_txn_projection(f.txn, NULL); + break; + case FATAL_PROJECTION_SENT_IDLE: + projection(&f, &p); + turn_txn_projection_sent(f.txn, &p); + break; + case FATAL_PROJECTION_SENT_NULL: + turn_txn_begin(f.txn); + turn_txn_projection_sent(f.txn, NULL); + break; + case FATAL_COMMIT_IDLE: + turn_txn_commit(f.txn, TURN_TXN_OK); + break; + case FATAL_COMMIT_OUTCOME: + turn_txn_begin(f.txn); + turn_txn_commit(f.txn, (enum turn_txn_commit)99); + break; + 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); + break; + case FATAL_CLEAR_ACTIVE: + turn_txn_begin(f.txn); + turn_txn_clear(f.txn, NULL); + break; + case FATAL_CLEAR_CONTEXT: + memset(&context, 0, sizeof(context)); + context.personal_len = 1; + turn_txn_clear(f.txn, &context); + break; + case FATAL_COMPACT_ACTIVE: + turn_txn_begin(f.txn); + turn_txn_compact(f.txn, "x", 1); + break; + case FATAL_COMPACT_NULL: + turn_txn_compact(f.txn, NULL, 1); + break; + case FATAL_FREE_ACTIVE: + turn_txn_begin(f.txn); + turn_txn_free(f.txn); + break; + case FATAL_NEW_NULL_SEED: + case FATAL_NEW_NULL_JOURNAL: + case FATAL_NEW_BAD_JOBS: + break; + } +} + +static int +case_is_fatal(enum fatal_case which) +{ + pid_t pid; + int status; + + fflush(NULL); + if ((pid = fork()) == -1) + return (0); + if (pid == 0) { + (void)close(STDERR_FILENO); + (void)open("/dev/null", O_WRONLY); + run_fatal_case(which); + _exit(0); + } + if (waitpid(pid, &status, 0) != pid) + return (0); + return (WIFEXITED(status) && WEXITSTATUS(status) == 1); +} + +static void +test_contract_failures_are_fail_closed(void) +{ + enum fatal_case which; + + for (which = FATAL_BEGIN_ACTIVE; which <= FATAL_NEW_BAD_JOBS; which++) + CHECK(case_is_fatal(which)); +} + +int +main(void) +{ + log_init(LOG_TO_STDERR, 0, LOG_USER); + test_success_is_durable_before_job_commit(); + test_abandon_aborts_jobs_then_rolls_back_pending(); + test_sent_snapshots_advance_and_stale_tokens_fail(); + test_foreign_snapshot_is_rejected(); + test_cap_does_not_claim_unsent_messages(); + test_resume_clear_compact_and_cache_reset_ownership(); + test_contract_failures_are_fail_closed(); + REGRESS_END(); +} blob - 4bd62d0c749d40a8da35020c4cbce134e1103af7 blob + 0c9c6f59381bc6cb457c42e3e72c9324fca043dc --- src/fugu/Makefile +++ src/fugu/Makefile @@ -1,7 +1,7 @@ PROG= fugu SRCS= main.c coord.c priv.c conf.c parse.y \ journal.c sysprompt.c tooldefs.c agentcfg.c skills.c output.c \ - anthropic_req.c openai_req.c generation.c msg.c \ + turn_txn.c anthropic_req.c openai_req.c generation.c msg.c \ json.c model_window.c buf.c imsgev.c log.c xmalloc.c MAN= fugu.1 fugu.conf.5 MANDIR= ${PREFIX}/man/man blob - /dev/null blob + cf405af16f58485c538794076e0a4edde1b491cb (mode 644) --- /dev/null +++ src/fugu/turn_txn.c @@ -0,0 +1,286 @@ +/* + * Copyright (c) 2026 Isaac + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +#include +#include + +#include +#include +#include + +#include "log.h" +#include "msg.h" +#include "journal.h" +#include "xmalloc.h" +#include "turn_txn.h" +#include "turn_txn_private.h" + +enum turn_txn_state { + TURN_TXN_IDLE, + TURN_TXN_ACTIVE +}; + +struct turn_txn { + struct msglist messages; + size_t message_count; + struct msg *pending_first; + size_t sent_prefix_len; + uint64_t snapshot_version; + int turn_n; + enum turn_txn_state state; + struct journal *journal; + struct turn_txn_jobs jobs; +}; + +static void +require_txn(const struct turn_txn *txn) +{ + if (txn == NULL) + fatalx("NULL Turn transaction"); +} + +static void +require_idle(const struct turn_txn *txn) +{ + require_txn(txn); + if (txn->state != TURN_TXN_IDLE) + fatalx("Turn transaction is active"); +} + +static void +require_active(const struct turn_txn *txn) +{ + require_txn(txn); + if (txn->state != TURN_TXN_ACTIVE) + fatalx("Turn transaction is idle"); +} + +static void +snapshot_changed(struct turn_txn *txn) +{ + if (txn->snapshot_version == UINT64_MAX) + fatalx("Turn transaction snapshot version exhausted"); + txn->snapshot_version++; +} + +static void +check_jobs(const struct turn_txn_jobs *jobs) +{ + if (jobs == NULL || jobs->begin == NULL || + jobs->stop_executor_if_live == NULL || jobs->commit == NULL || + jobs->abort == NULL) + fatalx("incomplete Turn transaction job Adapter"); +} + +struct turn_txn * +turn_txn_new(struct journal *journal, struct msglist *seed, + const struct turn_txn_jobs *jobs) +{ + struct turn_txn *txn; + struct msg *m; + size_t count = 0; + + if (journal == NULL) + fatalx("NULL Turn transaction Journal"); + if (seed == NULL) + fatalx("NULL Turn transaction seed"); + check_jobs(jobs); + TAILQ_FOREACH(m, seed, entry) { + if (count == INT_MAX) + fatalx("too many messages in Turn transaction seed"); + count++; + } + txn = xcalloc(1, sizeof(*txn)); + TAILQ_INIT(&txn->messages); + TAILQ_CONCAT(&txn->messages, seed, entry); + txn->message_count = count; + txn->snapshot_version = 1; + txn->journal = journal; + txn->jobs = *jobs; + return (txn); +} + +void +turn_txn_free(struct turn_txn *txn) +{ + if (txn == NULL) + return; + require_idle(txn); + msglist_free(&txn->messages); + free(txn); +} + +void +turn_txn_begin(struct turn_txn *txn) +{ + require_idle(txn); + if (txn->turn_n == INT_MAX) + fatalx("Turn transaction number exhausted"); + txn->turn_n++; + journal_turn_begin(txn->journal, txn->turn_n); + txn->jobs.begin(txn->jobs.arg); + txn->state = TURN_TXN_ACTIVE; + txn->pending_first = NULL; + snapshot_changed(txn); +} + +void +turn_txn_accept(struct turn_txn *txn, struct msg *message) +{ + require_active(txn); + if (message == NULL) + fatalx("NULL message accepted by Turn transaction"); + if (txn->message_count == INT_MAX) + fatalx("too many messages in Turn transaction"); + /* Journal ownership is established before live Projection visibility. */ + journal_message(txn->journal, message); + TAILQ_INSERT_TAIL(&txn->messages, message, entry); + if (txn->pending_first == NULL) + txn->pending_first = message; + txn->message_count++; + snapshot_changed(txn); +} + +void +turn_txn_projection(const struct turn_txn *txn, + struct turn_txn_projection *out) +{ + require_txn(txn); + if (out == NULL) + fatalx("NULL Turn transaction Projection output"); + out->messages = &txn->messages; + out->message_count = txn->message_count; + out->cache_tail_msg = txn->sent_prefix_len == 0 ? -1 : + (int)(txn->sent_prefix_len - 1); + out->token = txn->snapshot_version; +} + +void +turn_txn_projection_sent(struct turn_txn *txn, + const struct turn_txn_projection *projection) +{ + int cache_tail_msg; + + require_active(txn); + cache_tail_msg = txn->sent_prefix_len == 0 ? -1 : + (int)(txn->sent_prefix_len - 1); + if (projection == NULL || projection->messages != &txn->messages || + projection->token != txn->snapshot_version || + projection->message_count != txn->message_count || + projection->cache_tail_msg != cache_tail_msg) + fatalx("stale Turn transaction Projection snapshot"); + txn->sent_prefix_len = projection->message_count; + snapshot_changed(txn); +} + +void +turn_txn_commit(struct turn_txn *txn, enum turn_txn_commit outcome) +{ + enum turn_outcome journal_outcome; + + require_active(txn); + switch (outcome) { + case TURN_TXN_OK: + journal_outcome = TURN_OK; + break; + case TURN_TXN_CAP: + journal_outcome = TURN_CAP; + break; + default: + fatalx("invalid Turn transaction commit outcome"); + } + journal_turn_end(txn->journal, txn->turn_n, journal_outcome); + txn->jobs.commit(txn->jobs.arg); + txn->pending_first = NULL; + txn->state = TURN_TXN_IDLE; + snapshot_changed(txn); +} + +static void +discard_pending(struct turn_txn *txn) +{ + struct msg *m, *next; + + for (m = txn->pending_first; m != NULL; m = next) { + next = TAILQ_NEXT(m, entry); + TAILQ_REMOVE(&txn->messages, m, entry); + msg_free(m); + if (txn->message_count == 0) + fatalx("Turn transaction message count underflow"); + txn->message_count--; + } + txn->pending_first = NULL; +} + +void +turn_txn_abandon(struct turn_txn *txn) +{ + require_active(txn); + txn->jobs.stop_executor_if_live(txn->jobs.arg); + txn->jobs.abort(txn->jobs.arg); + journal_turn_end(txn->journal, txn->turn_n, TURN_ABANDONED); + discard_pending(txn); + txn->sent_prefix_len = 0; + txn->state = TURN_TXN_IDLE; + snapshot_changed(txn); +} + +void +turn_txn_cache_reset(struct turn_txn *txn) +{ + require_idle(txn); + txn->sent_prefix_len = 0; + snapshot_changed(txn); +} + +static void +reset_projection(struct turn_txn *txn) +{ + msglist_free(&txn->messages); + TAILQ_INIT(&txn->messages); + txn->message_count = 0; + txn->pending_first = NULL; + txn->sent_prefix_len = 0; + snapshot_changed(txn); +} + +void +turn_txn_clear(struct turn_txn *txn, const struct turn_txn_context *context) +{ + require_idle(txn); + if (context != NULL && + ((context->personal == NULL && context->personal_len != 0) || + (context->project == NULL && context->project_len != 0))) + fatalx("invalid Turn transaction context span"); + journal_clear(txn->journal); + if (context != NULL) + journal_context(txn->journal, context->personal, + context->personal_len, context->project, + context->project_len); + journal_sync(txn->journal); + reset_projection(txn); +} + +void +turn_txn_compact(struct turn_txn *txn, const void *summary, size_t len) +{ + require_idle(txn); + if (summary == NULL && len != 0) + fatalx("invalid Turn transaction compaction span"); + journal_compact(txn->journal, summary, len); + journal_sync(txn->journal); + reset_projection(txn); +} blob - /dev/null blob + 0d511e884776f1230014b121031ad3012c6b5294 (mode 644) --- /dev/null +++ src/fugu/turn_txn.h @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2026 Isaac + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +#ifndef TURN_TXN_H +#define TURN_TXN_H + +#include + +#include + +struct msg; +struct msglist; +struct turn_txn; + +enum turn_txn_commit { + TURN_TXN_OK, + TURN_TXN_CAP +}; + +/* + * A Projection is a borrowed, read-only live view, valid only until the next + * mutating transaction call. Serialize and send it synchronously, then + * acknowledge that unchanged receipt. The token identifies the exact view. + */ +struct turn_txn_projection { + const struct msglist *messages; + size_t message_count; + int cache_tail_msg; + uint64_t token; +}; + +struct turn_txn_context { + const char *personal; + size_t personal_len; + const char *project; + size_t project_len; +}; + +void turn_txn_free(struct turn_txn *); +void turn_txn_begin(struct turn_txn *); +/* Transfers message ownership exactly once. */ +void turn_txn_accept(struct turn_txn *, struct msg *); +void turn_txn_projection(const struct turn_txn *, + struct turn_txn_projection *); +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 *); +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); + +#endif /* TURN_TXN_H */ blob - /dev/null blob + 9ff7ee6946833439d824488da7bbd24055ff1d31 (mode 644) --- /dev/null +++ src/fugu/turn_txn_private.h @@ -0,0 +1,41 @@ +/* + * Copyright (c) 2026 Isaac + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +#ifndef TURN_TXN_PRIVATE_H +#define TURN_TXN_PRIVATE_H + +#include "msg.h" +#include "turn_txn.h" + +struct journal; + +/* Private remote-but-owned background-job Adapter. */ +struct turn_txn_jobs { + void *arg; + void (*begin)(void *); + void (*stop_executor_if_live)(void *); + void (*commit)(void *); + void (*abort)(void *); +}; + +/* + * Moves every message out of seed and leaves seed initialized and empty. + * The jobs table is copied; journal and jobs->arg remain borrowed until free. + */ +struct turn_txn *turn_txn_new(struct journal *, struct msglist *, + const struct turn_txn_jobs *); + +#endif /* TURN_TXN_PRIVATE_H */