commit - 91b632582edaf81718fbffc9193f46706b137bd2
commit + 6700e6d8bc89354c9145ee84ad4d8a3a666cc8c2
blob - 526917744516c4cc3b90e94fbe98a5f519ed063a
blob + 8684989960ba557e858f510879a21bf60bdc7263
--- CONTEXT.md
+++ CONTEXT.md
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)
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
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 <bsd.subdir.mk>
blob - 57c2435bf5f099f573b300ffbe187a7635b10368
blob + fe102d4dafbcf9bd82f474cd2053e352e8bcc847
--- regress/README
+++ regress/README
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);
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
+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 <bsd.regress.mk>
blob - /dev/null
blob + 8cdfe1972919af2a12f240b2233d7d637a1bc04b (mode 644)
--- /dev/null
+++ regress/turn_txn/turn_txn_journal_test.c
+/*
+ * Copyright (c) 2026 Isaac <isaac@itm.works>
+ *
+ * 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 <sys/queue.h>
+#include <sys/types.h>
+
+#include <err.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <syslog.h>
+
+#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
+/*
+ * Copyright (c) 2026 Isaac <isaac@itm.works>
+ *
+ * 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 <sys/queue.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+
+#include <fcntl.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <syslog.h>
+#include <unistd.h>
+
+#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
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
+/*
+ * Copyright (c) 2026 Isaac <isaac@itm.works>
+ *
+ * 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 <sys/queue.h>
+#include <sys/types.h>
+
+#include <limits.h>
+#include <stdint.h>
+#include <stdlib.h>
+
+#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
+/*
+ * Copyright (c) 2026 Isaac <isaac@itm.works>
+ *
+ * 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 <sys/types.h>
+
+#include <stdint.h>
+
+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
+/*
+ * Copyright (c) 2026 Isaac <isaac@itm.works>
+ *
+ * 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 */
