docs: begin non-turn-based surfaces protocol planning

.vscode/settings.json

@@ -14,5 +14,6 @@
   },
   "search.exclude": {
     ".repos/**": true
-  }
+  },
+  "js/ts.experimental.useTsgo": true
 }

AGENTS.md

@@ -167,9 +167,14 @@ whole branching model. See
   work dir (`~/.t3/compose-work`, not tmpfs `/tmp`) before install. See
   [docs/fork-stack.md](./docs/fork-stack.md) ("Integration overlay compose and lockfiles").
 
-## Pull requests (required handoff)
+## Pull requests (when publishing)
 
-When implementation work for a user request is done (code, docs, config — not pure Q&A):
+Do not commit, rebase, push, or open/update a PR merely because an edit is complete. Do those
+things only when the user explicitly requests publication or the specific version-control action,
+or when another workflow in this file explicitly requires it (for example, Discord-originated
+work). A request to change code, docs, or config does not by itself authorize publication.
+
+When publication or a PR handoff is in scope:
 
 1. **Commit** the changes on a feature branch cut from `fork/dev`.
 2. **Open or update a PR against `fork/dev`** before handing off — for every kind of work, including

apps/server/src/ntbs/ExchangeRepository.test.ts

@@ -0,0 +1,166 @@
+import { describe, expect, it } from "@effect/vitest";
+import { Effect, Exit } from "effect";
+import { MessageId, ProjectId, ThreadId } from "@t3tools/contracts";
+import {
+  ExchangeRepositoryError,
+  ExchangeRepository,
+  inMemoryExchangeRepository,
+} from "./ExchangeRepository.ts";
+import {
+  makeRequestClaimed,
+  toReplyPending,
+  toReplyPosted,
+  toThreadCreated,
+  toUndeliverable,
+} from "./exchange.ts";
+
+const makeExchange = (sourceUri: string, threadId: string) =>
+  makeRequestClaimed(
+    {
+      sourceUri,
+      snapshot: "request",
+      attachments: [],
+    },
+    {
+      projectId: ProjectId.make("project"),
+      startBranchName: "main",
+      startCommitSha: "start-commit-sha",
+      threadId: ThreadId.make(threadId),
+      userMessageId: MessageId.make(`message-${threadId}`),
+      worktreeBranchName: `branch-${threadId}`,
+    },
+  );
+
+describe("inMemoryExchangeRepository", () => {
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("allows the same sourceUri to replace its state", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const claimed = makeExchange("test://request/1", "thread-1");
+        const threadCreated = toThreadCreated(claimed);
+
+        yield* repository.upsert(claimed);
+        yield* repository.upsert(threadCreated);
+
+        expect(yield* repository.findBySourceUri(claimed.sourceUri)).toEqual(threadCreated);
+      }),
+    );
+  });
+
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("rejects a threadId already owned by another sourceUri", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const existing = makeExchange("test://request/1", "shared-thread");
+        const conflicting = makeExchange("test://request/2", "shared-thread");
+
+        yield* repository.upsert(existing);
+        const error = yield* Effect.flip(repository.upsert(conflicting));
+
+        expect(error).toBeInstanceOf(ExchangeRepositoryError);
+        expect(error.reason).toContain(existing.t3.threadId);
+        expect(yield* repository.findBySourceUri(existing.sourceUri)).toEqual(existing);
+        expect(yield* repository.findBySourceUri(conflicting.sourceUri)).toBeNull();
+      }),
+    );
+  });
+
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("finds an exchange by threadId", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const exchange = makeExchange("test://request/1", "thread-1");
+
+        yield* repository.upsert(exchange);
+
+        expect(yield* repository.findByThreadId(exchange.t3.threadId)).toEqual(exchange);
+        expect(yield* repository.findByThreadId(ThreadId.make("unknown-thread"))).toBeNull();
+      }),
+    );
+  });
+
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("finds only non-terminal exchanges", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const claimed = makeExchange("test://request/claimed", "thread-claimed");
+        const threadCreated = toThreadCreated(
+          makeExchange("test://request/thread-created", "thread-created"),
+        );
+        const replyPending = toReplyPending(
+          toThreadCreated(makeExchange("test://request/reply-pending", "thread-reply-pending")),
+          { type: "answer", text: "pending reply" },
+        );
+        const replyPosted = toReplyPosted(
+          toReplyPending(
+            toThreadCreated(makeExchange("test://request/reply-posted", "thread-reply-posted")),
+            { type: "answer", text: "posted reply" },
+          ),
+          "test://reply/posted",
+        );
+        const undeliverable = toUndeliverable(
+          toReplyPending(
+            toThreadCreated(makeExchange("test://request/undeliverable", "thread-undeliverable")),
+            { type: "failure", text: "undeliverable reply", cause: "undeliverable te dico" },
+          ),
+          { message: "platform rejected the reply" },
+        );
+
+        yield* Effect.forEach(
+          [claimed, threadCreated, replyPending, replyPosted, undeliverable],
+          repository.upsert,
+        );
+
+        const results = yield* repository.findNonTerminalExchanges;
+
+        expect(results).toHaveLength(3);
+        expect(results).toEqual(expect.arrayContaining([claimed, threadCreated, replyPending]));
+      }),
+    );
+  });
+
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("preserves existing records when a replacement has a conflicting threadId", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const first = makeExchange("test://request/1", "thread-1");
+        const second = makeExchange("test://request/2", "thread-2");
+        const conflictingReplacement = makeExchange("test://request/2", "thread-1");
+
+        yield* repository.upsert(first);
+        yield* repository.upsert(second);
+        yield* Effect.flip(repository.upsert(conflictingReplacement));
+
+        expect(yield* repository.findBySourceUri(first.sourceUri)).toEqual(first);
+        expect(yield* repository.findBySourceUri(second.sourceUri)).toEqual(second);
+        expect(yield* repository.findByThreadId(first.t3.threadId)).toEqual(first);
+        expect(yield* repository.findByThreadId(second.t3.threadId)).toEqual(second);
+      }),
+    );
+  });
+
+  it.layer(inMemoryExchangeRepository)((it) => {
+    it.effect("atomically rejects concurrent upserts with the same threadId", () =>
+      Effect.gen(function* () {
+        const repository = yield* ExchangeRepository;
+        const first = makeExchange("test://request/1", "shared-thread");
+        const second = makeExchange("test://request/2", "shared-thread");
+
+        const outcomes = yield* Effect.all(
+          [Effect.exit(repository.upsert(first)), Effect.exit(repository.upsert(second))],
+          { concurrency: "unbounded" },
+        );
+
+        expect(outcomes.filter(Exit.isSuccess)).toHaveLength(1);
+        expect(outcomes.filter(Exit.isFailure)).toHaveLength(1);
+
+        const stored = yield* Effect.all([
+          repository.findBySourceUri(first.sourceUri),
+          repository.findBySourceUri(second.sourceUri),
+        ]);
+
+        expect(stored.filter((state) => state !== null)).toHaveLength(1);
+      }),
+    );
+  });
+});

apps/server/src/ntbs/ExchangeRepository.ts

@@ -0,0 +1,111 @@
+/*
+ * Defines the repository for durable NTBS exchanges.
+ *
+ * An exchange links an admitted external-platform request to its planned T3
+ * work and tracks its progress through delivery of the eventual reply.
+ *
+ * The repository owns persistence, lookup, and recovery. Each stored exchange
+ * is identified by its `sourceUri`, while the processor decides how to handle
+ * duplicate requests. It does not communicate with T3 or the originating
+ * platform.
+ */
+import { Array, Effect, Context, Data, HashMap, Ref, Layer } from "effect";
+import { isNonTerminal, type Exchange, type NonTerminalExchange } from "./exchange.ts";
+import type { ThreadId } from "@t3tools/contracts";
+import { isSome } from "effect/Option";
+
+export class ExchangeRepositoryError extends Data.TaggedError("ExchangeRepositoryError")<{
+  readonly reason: string;
+  readonly cause: unknown;
+}> {}
+
+export interface ExchangeRepository {
+  readonly findBySourceUri: (
+    sourceUri: string,
+  ) => Effect.Effect<Exchange | null, ExchangeRepositoryError>;
+
+  readonly findByThreadId: (
+    threadId: ThreadId,
+  ) => Effect.Effect<Exchange | null, ExchangeRepositoryError>;
+
+  readonly findNonTerminalExchanges: Effect.Effect<
+    ReadonlyArray<NonTerminalExchange>,
+    ExchangeRepositoryError
+  >;
+
+  /** Inserts or replaces the exchange identified by its `sourceUri`. */
+  readonly upsert: (exchange: Exchange) => Effect.Effect<void, ExchangeRepositoryError>;
+}
+
+export const ExchangeRepository = Context.Service<ExchangeRepository>(
+  "t3code/ntbs/ExchangeRepository",
+);
+
+const inMemoryER: Effect.Effect<ExchangeRepository> = Effect.gen(function* () {
+  const exchanges: Ref.Ref<HashMap.HashMap<string, Exchange>> = yield* Ref.make(
+    HashMap.empty<string, Exchange>(),
+  );
+
+  const upsert = Effect.fn("ExchangeRepository.upsert")(function* (exchange: Exchange) {
+    // we return conflicting source Uri as the first argument
+    // in case we find that the same threadId belongs already to a different sourceUri
+    const conflictingSourceUri = yield* Ref.modify(exchanges, (map) => {
+      const conflict = HashMap.findFirst(
+        map,
+        (existing, sourceUri) =>
+          sourceUri !== exchange.sourceUri && existing.t3.threadId === exchange.t3.threadId,
+      );
+
+      return isSome(conflict)
+        ? [conflict.value[0], map]
+        : [null, HashMap.set(map, exchange.sourceUri, exchange)];
+    });
+
+    if (conflictingSourceUri !== null) {
+      return yield* new ExchangeRepositoryError({
+        reason: `Thread ${exchange.t3.threadId} already belongs to exchange ${conflictingSourceUri}`,
+        cause: {
+          threadId: exchange.t3.threadId,
+          existingSourceUri: conflictingSourceUri,
+          incomingSourceUri: exchange.sourceUri,
+        },
+      });
+    }
+  });
+
+  const findBySourceUri = (uri: string) =>
+    Ref.get(exchanges).pipe(
+      Effect.map((map) => HashMap.get(map, uri)),
+      Effect.map((o) => (isSome(o) ? o.value : null)),
+    );
+
+  const findByThreadId = (threadId: ThreadId) =>
+    Ref.get(exchanges).pipe(
+      Effect.map((map) => HashMap.filter(map, (val) => val.t3.threadId === threadId)),
+      // if we get more than one Exchange in the HashMap, something's wrong
+      Effect.andThen((map) =>
+        HashMap.size(map) > 1
+          ? new ExchangeRepositoryError({
+              reason: "Exchange Repository contains more than one entry for thredId: " + threadId,
+              cause: map,
+            })
+          : Effect.succeed(Array.fromIterable(HashMap.entries(map))).pipe(
+              Effect.map((arr) => (arr.length === 1 ? arr[0]![1] : null)),
+            ),
+      ),
+    );
+
+  const findNonTerminalExchanges = Ref.get(exchanges).pipe(
+    Effect.map((map) => Array.fromIterable(HashMap.entries(map))),
+    Effect.map((arr) =>
+      Array.filter(
+        arr.map((el) => el[1]),
+        isNonTerminal,
+      ),
+    ),
+  );
+
+  return { upsert, findBySourceUri, findByThreadId, findNonTerminalExchanges };
+});
+
+export const inMemoryExchangeRepository = Layer.effect(ExchangeRepository, inMemoryER);

apps/server/src/ntbs/adapter.ts

@@ -0,0 +1,56 @@
+import type { ReplyPending, ThreadCreated, UndeliverableCause } from "./exchange.ts";
+import { Context, Data, Effect } from "effect";
+
+/**
+ * A platform operation failed without establishing that reply delivery is
+ * permanently impossible. The processor may retry the operation later.
+ */
+export class AdapterError extends Data.TaggedError("AdapterError")<{
+  readonly reason: string;
+  readonly cause: unknown;
+}> {}
+
+/** The platform definitively rejected delivery of a pending reply. */
+export class ReplyRejected extends Data.TaggedError("ReplyRejected")<{
+  readonly cause: UndeliverableCause;
+}> {}
+
+/**
+ * Defines the platform-specific operations used by the shared NTBS processor.
+ *
+ * An adapter communicates with one originating platform. It posts
+ * acknowledgements and replies, and can discover whether a particular pending
+ * reply was already posted. It does not persist exchange state, create T3
+ * threads, or interpret T3 events.
+ */
+export interface NTBSAdapter {
+  /**
+   * Posts a best-effort working acknowledgement for an exchange whose T3
+   * thread now exists. The acknowledgement is not part of the durable exchange
+   * lifecycle and its platform identifier is not retained.
+   */
+  readonly acknowledge: (state: ThreadCreated) => Effect.Effect<void, AdapterError>;
+
+  /**
+   * Posts the exact reply stored in `state` to the destination identified by
+   * its `sourceUri`.
+   *
+   * Returns an adapter-encoded URI locating the posted reply. `ReplyRejected`
+   * means the platform definitively refused delivery; other failures remain
+   * retryable.
+   */
+  readonly postReply: (state: ReplyPending) => Effect.Effect<string, AdapterError | ReplyRejected>;
+
+  /**
+   * Searches for the exact pending reply in case it was posted before the
+   * corresponding `ReplyPosted` state could be persisted.
+   *
+   * Returns its adapter-encoded source URI when found, or `null` otherwise.
+   */
+  readonly findPostedReply: (state: ReplyPending) => Effect.Effect<string | null, AdapterError>;
+}
+
+/**
+ * One tag for every platform. A processor resolves its adapter from the context it is built in, so each one is given the implementation for its own platform.
+ */
+export const NTBSAdapter = Context.Service<NTBSAdapter>("t3code/ntbs/adapter");