fix(platform): chat stall guard, atomic turn claim, sharing lifecycle

docs/de/platform/projects/concepts.md

@@ -13,7 +13,7 @@ Ein Projekt ist die Einheit, zu der Tale greift, wenn ein Arbeitsvorhaben diesel
 
 ## Was ein Projekt besitzt
 
-**Chats**, die im Projekt gestartet werden, tragen seinen Kontext automatisch. Sie bleiben deine, bis du an einem Chat **Mit Projekt teilen** umlegst — der Chats-Tab teilt sich entsprechend in **Deine Chats** und **Mit Projekt geteilt**. Das Teilen eines Chats blendet deine persönlichen Erinnerungen und Anweisungen aus den Antworten aus, die andere Mitglieder sehen.
+**Chats**, die im Projekt gestartet werden, tragen seinen Kontext automatisch. Sie bleiben deine, bis du an einem Chat **Mit Projekt teilen** umlegst — der Chats-Tab teilt sich entsprechend in **Deine Chats** und **Mit Projekt geteilt**. Das Teilen eines Chats blendet deine persönlichen Erinnerungen und Anweisungen aus den Antworten aus, die andere Mitglieder sehen. Verschiebst du einen geteilten Chat in ein anderes Projekt — oder nimmst ihn aus seinem Projekt heraus —, endet das Teilen, damit ein neues Publikum ihn nicht stillschweigend erbt: Lege **Mit Projekt teilen** wieder um, wenn die Mitglieder des neuen Projekts ihn lesen sollen.
 
 **Anweisungen** sind Kontext, der für jeden Chat im Projekt gilt — die Rahmung, die Randbedingungen und das Vokabular der Arbeit —, damit niemand sie pro Chat neu einfügt.
 

docs/en/platform/projects/concepts.md

@@ -13,7 +13,7 @@ A project is the unit Tale reaches for when a body of work needs the same files,
 
 ## What a project owns
 
-**Chats** started inside the project carry its context automatically. They stay yours until you flip **Share with project** on a chat — the Chats tab splits into **Your chats** and **Shared with project** accordingly. Sharing a chat hides your personal memories and instructions from the responses other members see.
+**Chats** started inside the project carry its context automatically. They stay yours until you flip **Share with project** on a chat — the Chats tab splits into **Your chats** and **Shared with project** accordingly. Sharing a chat hides your personal memories and instructions from the responses other members see. Moving a shared chat to another project — or out of its project — ends the share, so a new audience never inherits it silently: switch **Share with project** back on if the new project's members should read it.
 
 **Instructions** are context that applies to every chat in the project — the framing, constraints, and vocabulary of the work — so nobody re-pastes them per chat.
 

docs/fr/platform/projects/concepts.md

@@ -13,7 +13,7 @@ Un projet est l’unité que Tale sort quand un chantier a besoin des mêmes fic
 
 ## Ce qu’un projet possède
 
-Les **chats** démarrés dans le projet portent son contexte automatiquement. Ils restent les tiens jusqu’à ce que tu actives **Partager avec le projet** sur un chat — l’onglet Chats se divise en **Tes chats** et **Partagés avec le projet** en conséquence. Partager un chat masque tes souvenirs et tes instructions personnels dans les réponses que voient les autres membres.
+Les **chats** démarrés dans le projet portent son contexte automatiquement. Ils restent les tiens jusqu’à ce que tu actives **Partager avec le projet** sur un chat — l’onglet Chats se divise en **Tes chats** et **Partagés avec le projet** en conséquence. Partager un chat masque tes souvenirs et tes instructions personnels dans les réponses que voient les autres membres. Déplacer un chat partagé vers un autre projet — ou le sortir de son projet — met fin au partage, pour qu’un nouveau public n’en hérite jamais en silence : réactive **Partager avec le projet** si les membres du nouveau projet doivent le lire.
 
 Les **instructions** sont du contexte qui s’applique à chaque chat du projet — le cadre, les contraintes et le vocabulaire du travail — pour que personne ne les recolle chat par chat.
 

services/platform/backend/core/chat/stream_stall.test.ts

@@ -0,0 +1,76 @@
+// @vitest-environment node
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import {
+  createStallGuard,
+  STREAM_STALL_TIMEOUT_MS,
+  stallMessage,
+} from './stream_stall';
+
+/**
+ * The stall guard is a SILENCE clock, not a deadline: activity restarts it,
+ * so a stream that keeps producing can run for any length of time, and only
+ * a provider that goes quiet for the whole window trips it.
+ */
+
+beforeEach(() => {
+  vi.useFakeTimers();
+});
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe('createStallGuard', () => {
+  it('never fires while activity keeps arriving, however long the stream runs', () => {
+    const guard = createStallGuard(1_000);
+    // Ten windows' worth of wall clock, touched just before each deadline.
+    for (let i = 0; i < 10; i++) {
+      vi.advanceTimersByTime(999);
+      guard.touch();
+    }
+    expect(guard.signal.aborted).toBe(false);
+    expect(guard.stalled).toBe(false);
+    guard.dispose();
+  });
+
+  it('fires once the provider has been silent for the whole window', () => {
+    const guard = createStallGuard(1_000);
+    vi.advanceTimersByTime(999);
+    expect(guard.signal.aborted).toBe(false);
+    vi.advanceTimersByTime(1);
+    expect(guard.signal.aborted).toBe(true);
+    expect(guard.stalled).toBe(true);
+    expect(guard.signal.reason).toBeInstanceOf(Error);
+    expect((guard.signal.reason as Error).message).toBe(stallMessage(1_000));
+  });
+
+  it('measures silence from the LAST byte, not from the request start', () => {
+    const guard = createStallGuard(1_000);
+    vi.advanceTimersByTime(800);
+    guard.touch();
+    // 1.6s since the start — past a fixed deadline, but only 0.8s of silence.
+    vi.advanceTimersByTime(800);
+    expect(guard.signal.aborted).toBe(false);
+    vi.advanceTimersByTime(200);
+    expect(guard.signal.aborted).toBe(true);
+  });
+
+  it('dispose stops the clock, and a late touch does not re-arm it', () => {
+    const guard = createStallGuard(1_000);
+    guard.dispose();
+    guard.touch();
+    vi.advanceTimersByTime(5_000);
+    expect(guard.signal.aborted).toBe(false);
+    expect(guard.stalled).toBe(false);
+  });
+
+  it('names the silence window and the timeout in the surfaced error', () => {
+    const guard = createStallGuard(STREAM_STALL_TIMEOUT_MS);
+    const error = guard.error(new Error('aborted'));
+    expect(error.message).toMatch(/timed out after 180 seconds of silence/);
+    expect(error.cause).toBeInstanceOf(Error);
+    guard.dispose();
+  });
+});

services/platform/backend/core/chat/stream_stall.ts

@@ -0,0 +1,79 @@
+/**
+ * The silence clock for a streaming model round.
+ *
+ * A model round used to run under ONE fixed wall-clock abort on the fetch —
+ * a cap that could not tell a reply still streaming healthily at minute four
+ * from a connection that died at minute one, so every long reply (a high
+ * reasoning effort, a large output ceiling) was cut mid-sentence at exactly
+ * the deadline and surfaced as a generic provider error. This guard measures
+ * SILENCE instead: its clock restarts on every byte the provider sends, so a
+ * stream that keeps producing is never aborted however long it runs, and only
+ * a provider that stops sending for the whole window ends the round. The
+ * first byte gets the same allowance — a slow-thinking model is not a hung
+ * one.
+ */
+
+/** How long the provider may stay silent — measured BETWEEN bytes, never
+ * from the request start — before the round is abandoned as stalled. */
+export const STREAM_STALL_TIMEOUT_MS = 180_000;
+
+export interface StallGuard {
+  /** Aborts once the provider has been silent for the whole window. Attach
+   * it to the fetch alongside the turn's own cancel signal. */
+  readonly signal: AbortSignal;
+  /** True once THIS guard aborted the signal — a stall, as opposed to a user
+   * cancel riding the same fetch. */
+  readonly stalled: boolean;
+  /** Bytes arrived: restart the silence clock. */
+  touch(): void;
+  /** The round ended (either way): stop the clock so nothing fires late. */
+  dispose(): void;
+  /** The failure to surface for a stall, in the user's face. "timed out"
+   * lands it in the chat-error classifier's transient bucket. */
+  error(cause?: unknown): Error;
+}
+
+export function stallMessage(timeoutMs: number): string {
+  return `The model provider stopped sending data — the reply timed out after ${Math.round(timeoutMs / 1000)} seconds of silence.`;
+}
+
+export function createStallGuard(
+  timeoutMs: number = STREAM_STALL_TIMEOUT_MS,
+): StallGuard {
+  const controller = new AbortController();
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let stalled = false;
+  let disposed = false;
+  const clear = (): void => {
+    if (timer !== undefined) clearTimeout(timer);
+    timer = undefined;
+  };
+  const error = (cause?: unknown): Error =>
+    new Error(
+      stallMessage(timeoutMs),
+      cause === undefined ? undefined : { cause },
+    );
+  const arm = (): void => {
+    clear();
+    timer = setTimeout(() => {
+      timer = undefined;
+      stalled = true;
+      controller.abort(error());
+    }, timeoutMs);
+  };
+  arm();
+  return {
+    signal: controller.signal,
+    get stalled() {
+      return stalled;
+    },
+    touch() {
+      if (!disposed && !stalled) arm();
+    },
+    dispose() {
+      disposed = true;
+      clear();
+    },
+    error,
+  };
+}

services/platform/backend/core/chat/turn_action.test.ts

@@ -0,0 +1,91 @@
+// @vitest-environment node
+
+import { describe, expect, it } from 'vitest';
+
+import { createStallGuard, type StallGuard } from './stream_stall';
+import { streamSse } from './turn_action';
+
+/**
+ * The provider stream's one clock is a silence clock: a reply that keeps
+ * arriving is never cut, however long it runs past what a fixed deadline
+ * would have allowed, and only a provider that stops sending ends the round
+ * — with a failure that names the stall, not a generic abort.
+ */
+
+function frame(text: string): string {
+  return `data: ${JSON.stringify({
+    choices: [{ index: 0, delta: { content: text }, finish_reason: null }],
+  })}\n\n`;
+}
+
+/** An SSE body that emits `count` frames `everyMs` apart and then closes —
+ * or, with `hang`, goes silent forever after the last one. */
+function drippingResponse(options: {
+  count: number;
+  everyMs: number;
+  hang?: boolean;
+}): Response {
+  const encoder = new TextEncoder();
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      let sent = 0;
+      const tick = (): void => {
+        sent += 1;
+        controller.enqueue(encoder.encode(frame(`tick${sent} `)));
+        if (sent < options.count) setTimeout(tick, options.everyMs);
+        else if (options.hang !== true) controller.close();
+      };
+      setTimeout(tick, options.everyMs);
+    },
+  });
+  return new Response(stream, {
+    headers: { 'content-type': 'text/event-stream' },
+  });
+}
+
+async function collect(response: Response, guard: StallGuard): Promise<string> {
+  const texts: string[] = [];
+  for await (const chunk of streamSse(response, 'openai', guard)) {
+    texts.push(chunk.text);
+  }
+  return texts.join('');
+}
+
+describe('streamSse under the stall guard', () => {
+  it('keeps a healthy stream alive far past the silence window', async () => {
+    const guard = createStallGuard(150);
+    // 30 frames 15ms apart: ~450ms of streaming against a 150ms window — a
+    // fixed deadline of the window's length would have cut this reply at
+    // frame ten.
+    const text = await collect(
+      drippingResponse({ count: 30, everyMs: 15 }),
+      guard,
+    );
+    guard.dispose();
+    expect(text.startsWith('tick1 tick2 ')).toBe(true);
+    expect(text.endsWith('tick30 ')).toBe(true);
+    expect(guard.stalled).toBe(false);
+  });
+
+  it('ends a stream whose provider goes silent, naming the stall', async () => {
+    const guard = createStallGuard(100);
+    await expect(
+      collect(drippingResponse({ count: 2, everyMs: 10, hang: true }), guard),
+    ).rejects.toThrow(/timed out after \d+ seconds of silence/);
+    guard.dispose();
+    expect(guard.stalled).toBe(true);
+  });
+
+  it('lets a user cancel riding the same fetch through as itself, not as a stall', async () => {
+    const guard = createStallGuard(1_000);
+    const abort = new DOMException('The operation was aborted.', 'AbortError');
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        setTimeout(() => controller.error(abort), 5);
+      },
+    });
+    await expect(collect(new Response(stream), guard)).rejects.toBe(abort);
+    guard.dispose();
+    expect(guard.stalled).toBe(false);
+  });
+});