Merge develop (durable turns, defaults, indexes, FKs) into feature/transcript-redaction

- Turn start keeps develop's supersede/concurrency, run tracker,
  register and forceFinalize, and attaches the TranscriptRedactor.
- The shared finalization flushes redaction after the final events and
  before the terminal turn update on every path (supersede, timeout,
  restart recovery, forceFinalize); the redactor always closes.
- Token-bearing columns widened to TEXT (H2-safe upgrade statements),
  entity length limits removed so tokens never truncate.
- Event reader race that could omit terminal events fixed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -1,5 +1,40 @@
 # Changelog
 
+## Unreleased - Transcript redaction
+
+- Tokenize classified PII/PHI in stored messages, tool inputs/results, replay events, pending actions and derived summaries using QQQ secured-data tokens. Preserve eligible model requests and live SSE.
+- Stage content until turn completion, redact split deltas and classify criteria even when no records match; protect assistant logs with a scoped core redactor.
+- Reject proposals whose parameters were redacted; prevent confirmation before turn completion.
+- Read core table-level classification metadata and add secured-data canary tests for storage, audits, logs and in-memory model access.
+- Preserve durable-turn watchdog, supersede, shutdown and restart recovery with turn-scoped transcript redaction; widen token-bearing columns and entity limits, with H2-safe upgrade SQL.
+- Require QQQ 4.1.0-RC.7 (qqq#1182); retain the default audit-history opt-in and avoid duplicated routing headers against current core. RC.7 is released.
+## Unreleased - Foreign keys
+
+- Reference DDL: add the missing foreign keys (conversation tree, usage, data audit,
+  budget accounting, profile skills) with orphan quarantine and their indexes; make
+  `assistant_model.provider_id` `ON DELETE CASCADE`. A Postgres guardrail test fails
+  any declared relation without a foreign key.
+
+## Unreleased - Durable turns
+
+- Renew user-owned run leases with heartbeats, poll cross-replica Stop, enforce
+  hard deadlines, and finalize active runs during graceful shutdown. Recover
+  abandoned runs on replay, active discovery, or the next question.
+- Default to superseding a prior answer in the same conversation. Accept explicit
+  concurrent questions; the per-user cap supersedes the oldest run instead of
+  refusing another question. Per-minute limits and budgets still apply.
+- Add `GET /qqq/v1/assistant/turns/active`, `turn.phase` progress and `startedAt`
+  on `message.start`. Cancel is idempotent and returns HTTP 200.
+- Add the optional `AssistantRunTracker` Jobs tray SPI (no-op by default),
+  pending QQQ core job tracking (#1210).
+- Add the durable-turn schema upgrade; apply it before deploying this runtime.
+  A partial unique index fences duplicate terminal events from concurrent reapers.
+- UI API extension `assistant` 1.2.0 (additive): `turns/active`, the optional
+  `concurrency` on a new turn, 503 `SHUTTING_DOWN`, `turn.phase`,
+  `message.start.startedAt`, cancelReason `SUPERSEDED`. The 429 response stays
+  documented for budgets and per-minute limits; the "answer already in progress"
+  refusal is no longer returned. Cancel keeps 202 documented for older replicas.
+
 ## Unreleased - Starter defaults
 
 - Every seed changeset's rollback is empty: a seeded row cannot be told apart from a host's own row with the same name, and Liquibase also rolls back changesets it marked ran, so a name-based undo would delete host data.

CLAUDE.md

@@ -52,6 +52,8 @@ Always use the wrapper: Maven 3.10 cannot read the parent POM. The committed def
 
 ## Invariants (do not break)
 
+- **Transcript redaction:** stored transcripts never contain raw PII/PHI values; tokens via `ClassifiedValueTokens`. Redact storage copies before truncation/encryption; eligible model inputs and live SSE remain unchanged.
+
 - **Sessions:**
   - Scheduled runs log in as the designated user through `POST /qqq/v1/manageSession`; resolve only its returned UUID, verify the user id, and never construct or elevate a session. The QQQ-supplied scheduler context touches only assistant bookkeeping.
   - Never use `QSystemUserSession` or construct a `QSession` in main code (ArchUnit).

README.md

@@ -30,6 +30,25 @@ Hosts register providers through `AssistantQBitConfig` in `qbit-assistant`.
 
 Data tools use the QQQ UI API (`/qqq/v1`) with the signed-in user's session. The UI API enforces permissions and record locks; assistant data-access policy and field exposure can only narrow access. Only record-free navigation and help use the core AI tool kernel. Every tool result reaches the model wrapped as untrusted data.
 
+## Transcript privacy
+
+QQQ **4.1.0-RC.7** supplies the secured-data tokens and log hook (`qqq#1182`).
+Stored transcripts never contain the raw PII/PHI values read by a turn or supplied
+in classified tool criteria/fields. Messages, tool logs, replay events, knowledge
+snippets and pending payloads store `ClassifiedValueTokens` tokens; scheduled
+summaries use the redacted message. The model and live user stream retain the
+values permitted by PHI routing. History and replay show tokens instead.
+
+Content is finalized at turn completion; interrupted servers leave value-free
+placeholders. Proposals requiring redacted parameters are rejected and must be
+completed in the application. Existing transcripts are not migrated. See
+[transcript redaction](docs/DESIGN.md#transcript-redaction-secured-data-15).
+The dependency upgrade stays draft until RC.7 is published; local verification:
+
+```powershell
+.\mvnw.cmd -q verify "-Dqqq.snapshot.version=4.1.0-RC7PRE-SNAPSHOT"
+```
+
 ## Host audit attribution (X1/Y1)
 
 Use QQQ **4.1.0-RC.6 or later**. The QBit explicitly denies every assistant

UPGRADE.md

@@ -1,5 +1,81 @@
 # Upgrading qbit-assistant
 
+## Transcript redaction (Unreleased)
+
+Apply the transcript-redaction column widening block in `docs/schema-postgres.sql`
+to existing databases. Stop reasons, request IDs, tool-call IDs, feedback/data-audit/
+budget-override reasons and scheduled-run summaries/errors now use `TEXT`; token
+expansion must not fail a write or truncate secured-data tokens. The corresponding
+entity field limits are removed, including pending-action and tool-result summaries.
+
+## Foreign keys (2026-10-14)
+
+Recommended: apply the "Foreign keys" block at the end of
+`docs/schema-postgres.sql` (after the index audit). It is guarded and safe to re-run.
+- Adds the missing foreign keys across the conversation tables, usage, data audit,
+  budget accounting and profile skills, with the indexes they need.
+- Detail rows of a conversation `CASCADE`; the message/turn cycle, usage and the data
+  audit use `SET NULL`, so usage keeps its cost rows when a conversation is deleted.
+- Usage, budget accounting and the data audit hold their provider (`NO ACTION`):
+  disable a provider with history instead of deleting it.
+- `assistant_model.provider_id` changes from `NO ACTION` to `CASCADE`
+  (constraint renamed `fk_assistant_model_provider`).
+- Rows that point at a missing parent are first copied to `qqq_fk_quarantine`
+  (created if absent), then nulled, or deleted for the `NOT NULL` profile-skill
+  links. Check that table after the upgrade.
+
+No code change.
+
+## Index audit (2026-10-14)
+
+Recommended, not required: apply the "Index audit" block at the end of
+`docs/schema-postgres.sql`. It adds six indexes, every one `IF NOT EXISTS`:
+- usage by turn (read on each scheduled run)
+- the three retention date filters (turn events, tool calls, inactive conversations)
+- the two `ON DELETE SET NULL` foreign keys on `assistant_scheduled_run`
+
+No code change.
+
+## Durable turns (Unreleased)
+
+Apply the **Durable turns** block at the end of `docs/schema-postgres.sql` before
+starting the new runtime. It adds nullable `heartbeat_at`, `lease_owner`,
+`cancel_requested_at`, `phase`, `concurrency_mode` and the `(user_id, status)`
+index. A partial unique index on terminal events also prevents two replicas from
+inserting separate `message.end` frames for the same turn. Check for historical
+duplicate terminal frames before applying it (resolve any duplicates through your
+normal migration process). Fresh installs run the full schema, including ALTERs.
+Existing rows with no heartbeat use `started_at` for lease expiry. During the
+first upgrade from a runtime without heartbeats, drain old replicas' turns first:
+a new replica cannot distinguish an old live turn from a dead owner after 45 s.
+
+Defaults: heartbeat 5 s, lease 45 s, hard duration 300 s, force-finalization grace
+20 s, shutdown grace 8 s. Configure these through the corresponding
+`AssistantQBitConfig` withers. Heartbeat must be shorter than the lease. The JVM
+shutdown hook is enabled by default; tests disable `registerShutdownHook` and
+managed hosts may call `runtime.shutdown(Duration)` themselves. Give the process
+termination budget enough time for the configured shutdown grace and persistence.
+
+`POST /conversations/{id}/turns` accepts `concurrency: "supersede" | "concurrent"`
+(case-insensitive). Default `SUPERSEDE` stops prior runs in that conversation;
+`CONCURRENT` permits them. The default per-user cap is now 3; reaching it stops
+the oldest run instead of returning the old in-progress 429. Per-minute and
+budget 429 responses still apply. Superseded turns record `cancelReason=SUPERSEDED`.
+
+Cancel now returns HTTP **200**, including for terminal turns, with the current
+`status`. The OpenAPI fragment retains the old 202 response for mixed-version
+rollouts. Active runs are discoverable through `GET /qqq/v1/assistant/turns/active`;
+reattach through the existing events route. Consumers should recognize
+`turn.phase` and `message.start.startedAt`. Starts during shutdown return
+503 `SHUTTING_DOWN`; clients may retry against a ready replica.
+
+Hosts may wire `AssistantRunTracker` for Jobs tray publication, progress, Stop,
+and terminal notification. It is a no-op by default and all callbacks are
+isolated from turn execution. Tracking ids are local to the owning runtime;
+reaping an already dead owner does not invent another tracking job. Hosts should
+associate their durable tracking row with the supplied turn id. The SPI is a
+stand-in for QQQ #1210.
+
 ## Starter defaults, setup card and table help
 
 The jar now ships `db/liquibase/qbit-assistant/assistant-defaults.changelog.sql`.

docs/DESIGN.md

@@ -637,3 +637,32 @@ The shared golden task asks "How many images do I have by type?": five UI count
 calls establish 81 JPEG, 20 PNG and zero other configured types, then a presentation
 with two rows, 80.2%/19.8% shares, two pie points and working MIME filter links.
 The same task is included in the opt-in `live` Anthropic eval suite.
+
+
+## Transcript redaction (secured data §15)
+
+`TranscriptRedactor` collects classified values from tools using `ClassificationSource`,
+including core table/field metadata. Stored record fields use
+`ClassifiedValueTokens.forRecordField`; criteria/proposals without a record use
+`forContext`. Core selects `shred:v1`, `hmac:v1`, or `changed:v1`. Knowledge snippets
+retain their source field's token. Matching is case-sensitive and includes canonical,
+displayed, escaped, encoded and explicitly truncated variants.
+
+`AssistantStore` stages content in turn memory and writes placeholders until
+finalization, protecting backend writes and their audits. It redacts before truncation
+and encryption. This covers messages, tool/bulk logs, replay events, pending payloads
+and audit reasons; scheduled summaries derive from redacted messages. Model requests
+and live SSE remain unchanged under PHI routing. Persisted text deltas are joined
+before redaction, with the text in the first delta and remaining deltas empty;
+sequence numbers remain stable.
+
+Crashes or unfinished tools at timeout leave placeholders. Proposals remain
+non-executable (`PREPARING`) until finalization; redacted proposals are rejected and
+must be completed in the application. A core `LogRedactorInterface` hook covers all
+threads until the turn and its tool workers finish, then releases the raw values.
+
+History stays redacted; later turns re-read records. Existing transcripts are not
+migrated, and arbitrary text is not inferred to be classified without a matching
+observed value or classified criterion/field. Canary tests scan every assistant table,
+core audits and captured logs, while proving eligible models receive original values.
+Requires QQQ 4.1.0-RC.7 (`qqq#1182`); keep the PR draft until publication.

docs/schema-postgres.sql

@@ -49,7 +49,7 @@ CREATE TABLE assistant_turn
    status VARCHAR(16),
    classification VARCHAR(16),
    classification_tables VARCHAR(500),
-   stop_reason VARCHAR(64),
+   stop_reason TEXT,
    error_code VARCHAR(32),
    cancel_reason VARCHAR(64),
    model VARCHAR(200),
@@ -88,7 +88,7 @@ CREATE TABLE assistant_tool_call
    turn_id INTEGER,
    conversation_id INTEGER,
    user_id VARCHAR(255),
-   tool_call_id VARCHAR(128),
+   tool_call_id TEXT,
    tool_name VARCHAR(64),
    input_json TEXT,
    outcome VARCHAR(16),
@@ -118,7 +118,7 @@ CREATE TABLE assistant_feedback
    conversation_id INTEGER,
    user_id VARCHAR(255),
    rating INTEGER,
-   reason VARCHAR(64),
+   reason TEXT,
    comment TEXT,
    create_date TIMESTAMP WITH TIME ZONE
 );
@@ -267,7 +267,7 @@ CREATE TABLE assistant_usage
    cache_read_tokens BIGINT,
    cache_write_tokens BIGINT,
    cost_micros BIGINT,
-   request_id VARCHAR(100),
+   request_id TEXT,
    create_date TIMESTAMP WITH TIME ZONE
 );
 CREATE INDEX assistant_usage_create_date ON assistant_usage (create_date);
@@ -310,7 +310,7 @@ CREATE TABLE assistant_pending_action
    conversation_id INTEGER,
    turn_id INTEGER,
    user_id VARCHAR(255),
-   tool_call_id VARCHAR(128),
+   tool_call_id TEXT,
    tool_name VARCHAR(64),
    ui_api_operation VARCHAR(255),
    table_name VARCHAR(128),
@@ -382,7 +382,7 @@ CREATE TABLE assistant_data_audit
    classification VARCHAR(16),
    tables VARCHAR(500),
    decision VARCHAR(16),
-   reason VARCHAR(1000),
+   reason TEXT,
    create_date TIMESTAMP WITH TIME ZONE
 );
 CREATE INDEX assistant_data_audit_user_id ON assistant_data_audit (user_id);
@@ -415,7 +415,7 @@ CREATE TABLE assistant_budget_override
    scope_value VARCHAR(200),
    extra_value NUMERIC,
    unlimited_until TIMESTAMP WITH TIME ZONE,
-   reason VARCHAR(1000) NOT NULL,
+   reason TEXT NOT NULL,
    granted_by VARCHAR(200),
    expires_at TIMESTAMP WITH TIME ZONE NOT NULL,
    create_date TIMESTAMP WITH TIME ZONE,
@@ -566,9 +566,9 @@ CREATE TABLE assistant_scheduled_run (
    status VARCHAR(20) NOT NULL CHECK (status IN ('RUNNING', 'OK', 'FAILED', 'SKIPPED', 'BLOCKED')),
    conversation_id INTEGER REFERENCES assistant_conversation(id) ON DELETE SET NULL,
    turn_id INTEGER REFERENCES assistant_turn(id) ON DELETE SET NULL,
-   summary VARCHAR(2000),
+   summary TEXT,
    cost_micros BIGINT,
-   error VARCHAR(500),
+   error TEXT,
    create_date TIMESTAMPTZ,
    modify_date TIMESTAMPTZ,
    UNIQUE (task_id, scheduled_for)
@@ -577,3 +577,139 @@ CREATE INDEX assistant_scheduled_run_owner ON assistant_scheduled_run(user_id, s
 
 -- X1/Y1: encoded audit filters can exceed the old 500-character record links.
 ALTER TABLE assistant_tool_call ALTER COLUMN audit_href TYPE TEXT;
+
+
+-- Durable turns: apply this additive block before deploying the new runtime.
+-- Like earlier additions, fresh installs run the CREATEs followed by ALTERs.
+ALTER TABLE assistant_turn ADD COLUMN heartbeat_at TIMESTAMP;
+ALTER TABLE assistant_turn ADD COLUMN lease_owner VARCHAR(100);
+ALTER TABLE assistant_turn ADD COLUMN cancel_requested_at TIMESTAMP;
+ALTER TABLE assistant_turn ADD COLUMN phase VARCHAR(64);
+ALTER TABLE assistant_turn ADD COLUMN concurrency_mode VARCHAR(16);
+CREATE INDEX assistant_turn_user_status ON assistant_turn (user_id, status);
+
+-- Fence concurrent orphan reapers at the database boundary. A turn ends once.
+-- Check for pre-existing duplicate message.end rows before applying this index.
+CREATE UNIQUE INDEX assistant_turn_event_terminal ON assistant_turn_event (turn_id)
+   WHERE event_type = 'message.end';
+
+-- Index audit (2026-10-14): lookup, retention and FK paths with no leading index.
+-- ScheduledTurn.cost reads usage by turn on every scheduled run.
+CREATE INDEX IF NOT EXISTS assistant_usage_turn ON assistant_usage (turn_id);
+-- Retention: deleteTurnEventsBefore, clearToolResultsBefore and listConversationsInactiveSince filter by date.
+CREATE INDEX IF NOT EXISTS assistant_turn_event_create_date ON assistant_turn_event (create_date);
+CREATE INDEX IF NOT EXISTS assistant_tool_call_create_date ON assistant_tool_call (create_date);
+CREATE INDEX IF NOT EXISTS assistant_conversation_modify_date ON assistant_conversation (modify_date);
+-- ON DELETE SET NULL children: deleting a conversation or turn would scan scheduled runs without these.
+CREATE INDEX IF NOT EXISTS assistant_scheduled_run_conversation ON assistant_scheduled_run (conversation_id);
+CREATE INDEX IF NOT EXISTS assistant_scheduled_run_turn ON assistant_scheduled_run (turn_id);
+
+-- Foreign keys (2026-10-14). Run on new and existing databases; every step is guarded
+-- and re-runnable. The rules follow how the qbit deletes:
+-- * AssistantStore.deleteConversation (also the retention purge) deletes turn events,
+--   tool calls, feedback, messages, turns, data audit and pending actions, then the
+--   conversation. The conversation's own detail rows CASCADE, so an admin delete of a
+--   conversation cleans up the same way.
+-- * The message <-> turn cycle (turn.user_message_id / assistant_message_id and
+--   message.turn_id) and the other optional pointers are SET NULL. Rows are inserted
+--   parent first, so nothing is deferred.
+-- * assistant_usage keeps its cost rows when a conversation goes (SET NULL), and the
+--   data audit trail is never cascaded into (SET NULL).
+-- * Provider references from usage, budget accounting and the data audit are NO ACTION
+--   (like assistant_turn.provider_id): disable a provider with history, do not delete
+--   it. A provider's model rows are its own (CASCADE).
+-- * assistant_budget_usage.usage_id has no foreign key on purpose: the projection
+--   outlives its usage row so budgets keep counting the spend.
+-- Rows that already point at a missing parent are recorded in qqq_fk_quarantine first,
+-- then nulled (nullable columns) or deleted with the full row recorded (NOT NULL).
+CREATE INDEX IF NOT EXISTS assistant_message_parent_message ON assistant_message (parent_message_id);
+CREATE INDEX IF NOT EXISTS assistant_message_turn ON assistant_message (turn_id);
+CREATE INDEX IF NOT EXISTS assistant_turn_user_message ON assistant_turn (user_message_id);
+CREATE INDEX IF NOT EXISTS assistant_turn_assistant_message ON assistant_turn (assistant_message_id);
+CREATE INDEX IF NOT EXISTS assistant_turn_provider ON assistant_turn (provider_id);
+CREATE INDEX IF NOT EXISTS assistant_tool_call_pending_action ON assistant_tool_call (pending_action_id);
+CREATE INDEX IF NOT EXISTS assistant_feedback_message ON assistant_feedback (message_id);
+CREATE INDEX IF NOT EXISTS assistant_pending_action_turn ON assistant_pending_action (turn_id);
+CREATE INDEX IF NOT EXISTS assistant_data_audit_provider ON assistant_data_audit (provider_id);
+CREATE INDEX IF NOT EXISTS assistant_budget_usage_provider ON assistant_budget_usage (provider_id);
+CREATE INDEX IF NOT EXISTS assistant_profile_skill_skill ON assistant_profile_skill (skill_id);
+CREATE INDEX IF NOT EXISTS assistant_budget_downgrade_model ON assistant_budget (downgrade_model_id);
+CREATE INDEX IF NOT EXISTS assistant_conversation_scheduled_task ON assistant_conversation (scheduled_task_id);
+CREATE INDEX IF NOT EXISTS assistant_model_fallback_model ON assistant_model (fallback_model_id);
+CREATE INDEX IF NOT EXISTS assistant_profile_simple_model ON assistant_profile (simple_model_id);
+CREATE INDEX IF NOT EXISTS assistant_profile_complex_model ON assistant_profile (complex_model_id);
+CREATE INDEX IF NOT EXISTS assistant_profile_classifier_model ON assistant_profile (classifier_model_id);
+CREATE INDEX IF NOT EXISTS assistant_rag_source_embedding_model ON assistant_rag_source (embedding_model_id);
+CREATE INDEX IF NOT EXISTS assistant_scheduled_task_profile ON assistant_scheduled_task (profile_id);
+
+CREATE TABLE IF NOT EXISTS qqq_fk_quarantine (id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, table_name VARCHAR(100) NOT NULL, column_name VARCHAR(100) NOT NULL, row_id VARCHAR(100), missing_parent_id VARCHAR(100), action VARCHAR(20) NOT NULL, row_data TEXT, changeset VARCHAR(200), quarantined_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);
+
+DO $$
+DECLARE
+   r   RECORD;
+   old TEXT;
+BEGIN
+   FOR r IN SELECT * FROM (VALUES
+      ('assistant_message',        'conversation_id',      'assistant_conversation',   'CASCADE',   TRUE),
+      ('assistant_message',        'parent_message_id',    'assistant_message',        'SET NULL',  TRUE),
+      ('assistant_message',        'turn_id',              'assistant_turn',           'SET NULL',  TRUE),
+      ('assistant_turn',           'conversation_id',      'assistant_conversation',   'CASCADE',   TRUE),
+      ('assistant_turn',           'user_message_id',      'assistant_message',        'SET NULL',  TRUE),
+      ('assistant_turn',           'assistant_message_id', 'assistant_message',        'SET NULL',  TRUE),
+      ('assistant_turn_event',     'turn_id',              'assistant_turn',           'CASCADE',   TRUE),
+      ('assistant_tool_call',      'conversation_id',      'assistant_conversation',   'CASCADE',   TRUE),
+      ('assistant_tool_call',      'turn_id',              'assistant_turn',           'CASCADE',   TRUE),
+      ('assistant_tool_call',      'pending_action_id',    'assistant_pending_action', 'SET NULL',  TRUE),
+      ('assistant_feedback',       'conversation_id',      'assistant_conversation',   'CASCADE',   TRUE),
+      ('assistant_feedback',       'message_id',           'assistant_message',        'CASCADE',   TRUE),
+      ('assistant_pending_action', 'conversation_id',      'assistant_conversation',   'CASCADE',   TRUE),
+      ('assistant_pending_action', 'turn_id',              'assistant_turn',           'SET NULL',  TRUE),
+      ('assistant_data_audit',     'conversation_id',      'assistant_conversation',   'SET NULL',  TRUE),
+      ('assistant_data_audit',     'turn_id',              'assistant_turn',           'SET NULL',  TRUE),
+      ('assistant_usage',          'conversation_id',      'assistant_conversation',   'SET NULL',  TRUE),
+      ('assistant_usage',          'turn_id',              'assistant_turn',           'SET NULL',  TRUE),
+      ('assistant_usage',          'provider_id',          'assistant_provider',       'NO ACTION', TRUE),
+      ('assistant_data_audit',     'provider_id',          'assistant_provider',       'NO ACTION', TRUE),
+      ('assistant_budget_usage',   'provider_id',          'assistant_provider',       'NO ACTION', TRUE),
+      ('assistant_profile_skill',  'profile_id',           'assistant_profile',        'CASCADE',   FALSE),
+      ('assistant_profile_skill',  'skill_id',             'assistant_skill',          'CASCADE',   FALSE)
+   ) AS v(child, col, parent, rule, nullable)
+   LOOP
+      IF NOT EXISTS (SELECT 1 FROM information_schema.table_constraints WHERE constraint_schema = current_schema() AND LOWER(constraint_name) = 'fk_' || r.child || '_' || r.col) THEN
+         IF r.nullable THEN
+            EXECUTE format('INSERT INTO qqq_fk_quarantine (table_name, column_name, row_id, missing_parent_id, action, changeset) '
+               || 'SELECT %L, %L, c.id::text, c.%I::text, ''NULLED'', ''qbit-assistant 2026-10-14 foreign keys'' FROM %I c '
+               || 'WHERE c.%I IS NOT NULL AND NOT EXISTS (SELECT 1 FROM %I p WHERE p.id = c.%I)', r.child, r.col, r.col, r.child, r.col, r.parent, r.col);
+            EXECUTE format('UPDATE %I c SET %I = NULL WHERE c.%I IS NOT NULL AND NOT EXISTS (SELECT 1 FROM %I p WHERE p.id = c.%I)', r.child, r.col, r.col, r.parent, r.col);
+         ELSE
+            EXECUTE format('INSERT INTO qqq_fk_quarantine (table_name, column_name, row_id, missing_parent_id, action, row_data, changeset) '
+               || 'SELECT %L, %L, c.id::text, c.%I::text, ''DELETED'', row_to_json(c)::text, ''qbit-assistant 2026-10-14 foreign keys'' FROM %I c '
+               || 'WHERE NOT EXISTS (SELECT 1 FROM %I p WHERE p.id = c.%I)', r.child, r.col, r.col, r.child, r.parent, r.col);
+            EXECUTE format('DELETE FROM %I c WHERE NOT EXISTS (SELECT 1 FROM %I p WHERE p.id = c.%I)', r.child, r.parent, r.col);
+         END IF;
+         EXECUTE format('ALTER TABLE %I ADD CONSTRAINT %I FOREIGN KEY (%I) REFERENCES %I (id) ON DELETE %s', r.child, 'fk_' || r.child || '_' || r.col, r.col, r.parent, r.rule);
+      END IF;
+   END LOOP;
+
+   -- assistant_model.provider_id: the inline NO ACTION key becomes CASCADE, named as on kof22.
+   IF NOT EXISTS (SELECT 1 FROM information_schema.table_constraints WHERE constraint_schema = current_schema() AND LOWER(constraint_name) = 'fk_assistant_model_provider') THEN
+      FOR old IN SELECT conname FROM pg_constraint
+         WHERE contype = 'f' AND conrelid = 'assistant_model'::regclass AND confrelid = 'assistant_provider'::regclass
+      LOOP
+         EXECUTE format('ALTER TABLE assistant_model DROP CONSTRAINT %I', old);
+      END LOOP;
+      ALTER TABLE assistant_model ADD CONSTRAINT fk_assistant_model_provider FOREIGN KEY (provider_id) REFERENCES assistant_provider (id) ON DELETE CASCADE;
+   END IF;
+END $$;
+
+-- Transcript redaction: tokens can expand beyond the original payload bounds.
+-- Apply to existing databases as well; SET DATA TYPE is also valid in H2.
+ALTER TABLE assistant_turn ALTER COLUMN stop_reason SET DATA TYPE TEXT;
+ALTER TABLE assistant_tool_call ALTER COLUMN tool_call_id SET DATA TYPE TEXT;
+ALTER TABLE assistant_feedback ALTER COLUMN reason SET DATA TYPE TEXT;
+ALTER TABLE assistant_usage ALTER COLUMN request_id SET DATA TYPE TEXT;
+ALTER TABLE assistant_pending_action ALTER COLUMN tool_call_id SET DATA TYPE TEXT;
+ALTER TABLE assistant_data_audit ALTER COLUMN reason SET DATA TYPE TEXT;
+ALTER TABLE assistant_budget_override ALTER COLUMN reason SET DATA TYPE TEXT;
+ALTER TABLE assistant_scheduled_run ALTER COLUMN summary SET DATA TYPE TEXT;
+ALTER TABLE assistant_scheduled_run ALTER COLUMN error SET DATA TYPE TEXT;

pom.xml

@@ -215,7 +215,8 @@
             <activeByDefault>true</activeByDefault>
          </activation>
          <properties>
-            <qqq.snapshot.version>4.1.0-RC.6</qqq.snapshot.version>
+            <!-- RC.7 carries qqq#1182 classified-value tokens and log redaction. Keep the PR draft until published. -->
+            <qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
          </properties>
          <dependencyManagement>
             <dependencies>

qbit-assistant/pom.xml

@@ -87,6 +87,25 @@
          <version>4.33.0</version>
          <scope>test</scope>
       </dependency>
+      <!-- the reference DDL guardrail (SchemaForeignKeyPostgresTest) -->
+      <dependency>
+         <groupId>org.testcontainers</groupId>
+         <artifactId>junit-jupiter</artifactId>
+         <version>1.21.4</version>
+         <scope>test</scope>
+      </dependency>
+      <dependency>
+         <groupId>org.testcontainers</groupId>
+         <artifactId>postgresql</artifactId>
+         <version>1.21.4</version>
+         <scope>test</scope>
+      </dependency>
+      <dependency>
+         <groupId>org.postgresql</groupId>
+         <artifactId>postgresql</artifactId>
+         <version>42.7.14</version>
+         <scope>test</scope>
+      </dependency>
    </dependencies>
 
    <build>

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitConfig.java

@@ -17,6 +17,7 @@
 package com.kingsrook.qbits.assistant;
 
 
+import javax.net.ssl.SSLContext;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -31,6 +32,7 @@ import java.util.regex.Pattern;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.kingsrook.qbits.assistant.classification.ClassificationSource;
 import com.kingsrook.qbits.assistant.classification.ConfiguredClassificationSource;
+import com.kingsrook.qbits.assistant.jobs.AssistantRunTracker;
 import com.kingsrook.qbits.assistant.rag.VectorStore;
 import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
 import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
@@ -44,7 +46,6 @@ import com.kingsrook.qqq.backend.core.model.metadata.producers.MetaDataCustomize
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfig;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 import com.kingsrook.qqq.backend.core.utils.StringUtils;
-import javax.net.ssl.SSLContext;
 
 
 /*******************************************************************************
@@ -99,12 +100,20 @@ public class AssistantQBitConfig implements QBitConfig
    private Integer               maxParallelToolCalls       = 4;
    private Integer               maxRowsToModel             = 50;
    private Integer               maxCellChars               = 500;
+   private Integer turnHeartbeatSeconds = 5;
+   private Integer turnLeaseSeconds = 45;
+   private Integer turnForceFinalizeGraceSeconds = 20;
+   private Integer shutdownGraceSeconds = 8;
+   private String defaultConcurrencyMode = "SUPERSEDE";
+   private Boolean registerShutdownHook = true;
+   private AssistantRunTracker runTracker = AssistantRunTracker.NOOP;
+
    private Integer               maxTurnSeconds             = 300;
    private Integer               maxUserMessageChars        = 8000;
    private Integer               historyMessageLimit        = 40;
    private AssistantBudgets      budgets                    = new AssistantBudgets();
    private Integer               maxTurnsPerUserPerMinute   = 10;
-   private Integer               maxConcurrentTurnsPerUser  = 1;
+   private Integer               maxConcurrentTurnsPerUser  = 3;
    private Integer               conversationRetentionDays  = 90;
    private Integer               toolResultRetentionDays    = 30;
    private Integer               approvalRetentionDays      = 365;
@@ -206,6 +215,23 @@ public class AssistantQBitConfig implements QBitConfig
       requirePositive(errors, "maxRowsToModel", maxRowsToModel);
       requirePositive(errors, "maxCellChars", maxCellChars);
       requirePositive(errors, "maxTurnSeconds", maxTurnSeconds);
+      requirePositive(errors, "turnHeartbeatSeconds", turnHeartbeatSeconds);
+      requirePositive(errors, "turnLeaseSeconds", turnLeaseSeconds);
+      requirePositive(errors, "turnForceFinalizeGraceSeconds", turnForceFinalizeGraceSeconds);
+      requirePositive(errors, "shutdownGraceSeconds", shutdownGraceSeconds);
+      if(turnHeartbeatSeconds != null && turnLeaseSeconds != null && turnLeaseSeconds < 2 * turnHeartbeatSeconds)
+      {
+         errors.add("turnLeaseSeconds must be at least twice turnHeartbeatSeconds.");
+      }
+      if(!"SUPERSEDE".equals(defaultConcurrencyMode) && !"CONCURRENT".equals(defaultConcurrencyMode))
+      {
+         errors.add("defaultConcurrencyMode must be SUPERSEDE or CONCURRENT.");
+      }
+      if(turnHeartbeatSeconds != null && turnLeaseSeconds != null && turnHeartbeatSeconds >= turnLeaseSeconds)
+      {
+         errors.add("turnHeartbeatSeconds must be less than turnLeaseSeconds.");
+      }
+
       requirePositive(errors, "maxUserMessageChars", maxUserMessageChars);
       requirePositive(errors, "historyMessageLimit", historyMessageLimit);
       requirePositive(errors, "maxTurnsPerUserPerMinute", maxTurnsPerUserPerMinute);
@@ -2008,4 +2034,223 @@ public class AssistantQBitConfig implements QBitConfig
       this.scheduledTasksSchedulerName = scheduledTasksSchedulerName;
       return (this);
    }
+
+
+
+   /*******************************************************************************
+    ** Getter for turnHeartbeatSeconds
+    *******************************************************************************/
+   public Integer getTurnHeartbeatSeconds()
+   {
+      return (this.turnHeartbeatSeconds);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for turnHeartbeatSeconds
+    *******************************************************************************/
+   public void setTurnHeartbeatSeconds(Integer turnHeartbeatSeconds)
+   {
+      this.turnHeartbeatSeconds = turnHeartbeatSeconds;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for turnHeartbeatSeconds
+    *******************************************************************************/
+   public AssistantQBitConfig withTurnHeartbeatSeconds(Integer turnHeartbeatSeconds)
+   {
+      this.turnHeartbeatSeconds = turnHeartbeatSeconds;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for turnLeaseSeconds
+    *******************************************************************************/
+   public Integer getTurnLeaseSeconds()
+   {
+      return (this.turnLeaseSeconds);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for turnLeaseSeconds
+    *******************************************************************************/
+   public void setTurnLeaseSeconds(Integer turnLeaseSeconds)
+   {
+      this.turnLeaseSeconds = turnLeaseSeconds;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for turnLeaseSeconds
+    *******************************************************************************/
+   public AssistantQBitConfig withTurnLeaseSeconds(Integer turnLeaseSeconds)
+   {
+      this.turnLeaseSeconds = turnLeaseSeconds;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for turnForceFinalizeGraceSeconds
+    *******************************************************************************/
+   public Integer getTurnForceFinalizeGraceSeconds()
+   {
+      return (this.turnForceFinalizeGraceSeconds);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for turnForceFinalizeGraceSeconds
+    *******************************************************************************/
+   public void setTurnForceFinalizeGraceSeconds(Integer turnForceFinalizeGraceSeconds)
+   {
+      this.turnForceFinalizeGraceSeconds = turnForceFinalizeGraceSeconds;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for turnForceFinalizeGraceSeconds
+    *******************************************************************************/
+   public AssistantQBitConfig withTurnForceFinalizeGraceSeconds(Integer turnForceFinalizeGraceSeconds)
+   {
+      this.turnForceFinalizeGraceSeconds = turnForceFinalizeGraceSeconds;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for shutdownGraceSeconds
+    *******************************************************************************/
+   public Integer getShutdownGraceSeconds()
+   {
+      return (this.shutdownGraceSeconds);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for shutdownGraceSeconds
+    *******************************************************************************/
+   public void setShutdownGraceSeconds(Integer shutdownGraceSeconds)
+   {
+      this.shutdownGraceSeconds = shutdownGraceSeconds;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for shutdownGraceSeconds
+    *******************************************************************************/
+   public AssistantQBitConfig withShutdownGraceSeconds(Integer shutdownGraceSeconds)
+   {
+      this.shutdownGraceSeconds = shutdownGraceSeconds;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for defaultConcurrencyMode
+    *******************************************************************************/
+   public String getDefaultConcurrencyMode()
+   {
+      return (this.defaultConcurrencyMode);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for defaultConcurrencyMode
+    *******************************************************************************/
+   public void setDefaultConcurrencyMode(String defaultConcurrencyMode)
+   {
+      this.defaultConcurrencyMode = defaultConcurrencyMode;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for defaultConcurrencyMode
+    *******************************************************************************/
+   public AssistantQBitConfig withDefaultConcurrencyMode(String defaultConcurrencyMode)
+   {
+      this.defaultConcurrencyMode = defaultConcurrencyMode;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for registerShutdownHook
+    *******************************************************************************/
+   public Boolean getRegisterShutdownHook()
+   {
+      return (this.registerShutdownHook);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for registerShutdownHook
+    *******************************************************************************/
+   public void setRegisterShutdownHook(Boolean registerShutdownHook)
+   {
+      this.registerShutdownHook = registerShutdownHook;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for registerShutdownHook
+    *******************************************************************************/
+   public AssistantQBitConfig withRegisterShutdownHook(Boolean registerShutdownHook)
+   {
+      this.registerShutdownHook = registerShutdownHook;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for runTracker
+    *******************************************************************************/
+   @JsonIgnore
+   public AssistantRunTracker getRunTracker()
+   {
+      return (this.runTracker == null ? AssistantRunTracker.NOOP : this.runTracker);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for runTracker
+    *******************************************************************************/
+   public void setRunTracker(AssistantRunTracker runTracker)
+   {
+      this.runTracker = runTracker;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for runTracker
+    *******************************************************************************/
+   public AssistantQBitConfig withRunTracker(AssistantRunTracker runTracker)
+   {
+      this.runTracker = runTracker;
+      return (this);
+   }
+
 }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantQBitProducer.java

@@ -211,6 +211,10 @@ public class AssistantQBitProducer implements QBitMetaDataProducer<AssistantQBit
       {
          enableRecordHistory(ClientChannelsMetaData.ofOrWithNew(qInstance));
       }
+      else
+      {
+         ClientChannelsMetaData.ofOrWithNew(qInstance).withDeniedPattern("assistant", "audit*");
+      }
       postProduceActions(new MetaDataProducerMultiOutput(), qInstance);
       return (new MetaDataProducerMultiOutput());
    }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/AssistantRuntime.java

@@ -17,8 +17,11 @@
 package com.kingsrook.qbits.assistant;
 
 
+import java.net.InetAddress;
 import java.time.Clock;
+import java.time.Duration;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -50,6 +53,7 @@ import com.kingsrook.qqq.backend.core.actions.ai.AiToolKernelSettings;
 import com.kingsrook.qqq.backend.core.actions.ai.exposure.AiExposurePolicy;
 import com.kingsrook.qqq.backend.core.actions.ai.output.AiRedactor;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qqq.backend.core.utils.StringUtils;
@@ -62,6 +66,11 @@ import com.kingsrook.qqq.backend.core.utils.StringUtils;
  *******************************************************************************/
 public class AssistantRuntime
 {
+   private static final QLogger LOG = QLogger.getLogger(AssistantRuntime.class);
+   private final String instanceId = instanceId();
+   private volatile boolean shuttingDown;
+   private Thread shutdownHook;
+
    private final ScheduledTurn scheduledTurn = new ScheduledTurn();
 
    private final RagService ragService;
@@ -103,12 +112,22 @@ public class AssistantRuntime
     **
     *******************************************************************************/
    public AssistantRuntime(QInstance qInstance, AssistantQBitConfig config, Clock clock)
+   {
+      this(qInstance, config, clock, new AssistantStore(config));
+   }
+
+
+
+   /*******************************************************************************
+    ** Explicit persistence collaborator for embedded runtimes and fault tests.
+    *******************************************************************************/
+   public AssistantRuntime(QInstance qInstance, AssistantQBitConfig config, Clock clock, AssistantStore store)
    {
       AssistantChannelWall.validate(qInstance);
       this.qInstance = qInstance;
       this.config = config;
       this.clock = clock;
-      this.store = new AssistantStore(config);
+      this.store = store;
       this.policyService = new PolicyService(config, store, clock);
       this.providerService = new ProviderService(config, store, clock);
       this.budgetService = new BudgetService(store, providerService, clock);
@@ -157,6 +176,21 @@ public class AssistantRuntime
       this.knowledgeTool = new KnowledgeTool(this, ragService, vectors);
       this.bulkLoadTools = new BulkLoadTools(qInstance, dataAccessService, config.getBulkLoadRunner());
       this.contextBuilder = new ContextBuilder(config, kernel, toolbox).withKnowledgeTool(knowledgeTool).withBulkLoadTools(bulkLoadTools);
+      if(Boolean.TRUE.equals(config.getRegisterShutdownHook()))
+      {
+         shutdownHook = new Thread(() ->
+         {
+            try
+            {
+               shutdown(Duration.ofSeconds(config.getShutdownGraceSeconds()));
+            }
+            catch(Throwable ignored)
+            {
+               // Shutdown must never fail the JVM hook.
+            }
+         }, "assistant-shutdown-" + instanceId);
+         Runtime.getRuntime().addShutdownHook(shutdownHook);
+      }
    }
 
 
@@ -312,8 +346,13 @@ public class AssistantRuntime
    /*******************************************************************************
     ** Track a turn running here.
     *******************************************************************************/
-   public RunningTurn register(RunningTurn turn)
+   public synchronized RunningTurn register(RunningTurn turn)
    {
+      if(shuttingDown)
+      {
+         turn.forceFinalize("SHUTDOWN");
+         return (turn);
+      }
       runningTurns.put(turn.getTurnId(), turn);
       return (turn);
    }
@@ -345,7 +384,17 @@ public class AssistantRuntime
     *******************************************************************************/
    public void cancelAll(String reason)
    {
-      runningTurns.values().forEach(t -> t.cancel(reason));
+      runningTurns.values().forEach(t ->
+      {
+         try
+         {
+            t.cancel(reason);
+         }
+         catch(Exception e)
+         {
+            LOG.warn("Assistant cancellation failed; finalization will still run");
+         }
+      });
    }
 
 
@@ -355,8 +404,7 @@ public class AssistantRuntime
     *******************************************************************************/
    public void close()
    {
-      runningTurns.values().forEach(t -> t.cancel("SHUTDOWN"));
-      executor.shutdownNow();
+      shutdown(Duration.ZERO);
    }
 
 
@@ -504,4 +552,97 @@ public class AssistantRuntime
       return (bulkLoadTools);
    }
 
+
+
+
+   /*******************************************************************************
+    ** Stop accepting work, cancel, wait a bounded grace, then force finalization.
+    *******************************************************************************/
+   public void shutdown(Duration grace)
+   {
+      synchronized(this)
+      {
+         shuttingDown = true;
+      }
+      cancelAll("SHUTDOWN");
+      long until = System.nanoTime() + Math.max(0, grace.toNanos());
+      while(!runningTurns.isEmpty() && System.nanoTime() < until)
+      {
+         try
+         {
+            Thread.sleep(10);
+         }
+         catch(InterruptedException e)
+         {
+            Thread.currentThread().interrupt();
+            break;
+         }
+      }
+      runningTurns.values().forEach(t ->
+      {
+         try
+         {
+            t.forceFinalize("SHUTDOWN");
+         }
+         catch(Exception e)
+         {
+            LOG.warn("Assistant shutdown finalization failed");
+         }
+      });
+      executor.shutdownNow();
+      if(shutdownHook != null && Thread.currentThread() != shutdownHook)
+      {
+         try
+         {
+            Runtime.getRuntime().removeShutdownHook(shutdownHook);
+         }
+         catch(IllegalStateException ignored)
+         {
+            // The JVM is already shutting down.
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for shuttingDown
+    *******************************************************************************/
+   public boolean isShuttingDown()
+   {
+      return (shuttingDown);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for instanceId
+    *******************************************************************************/
+   public String getInstanceId()
+   {
+      return (instanceId);
+   }
+
+
+
+   /*******************************************************************************
+    ** A stable, bounded hostname plus a per-runtime random suffix.
+    *******************************************************************************/
+   private static String instanceId()
+   {
+      String host = System.getenv("HOSTNAME");
+      if(!StringUtils.hasContent(host))
+      {
+         try
+         {
+            host = InetAddress.getLocalHost().getHostName();
+         }
+         catch(Exception ignored)
+         {
+            host = "unknown";
+         }
+      }
+      return (host.substring(0, Math.min(87, host.length())) + ":" + UUID.randomUUID().toString().substring(0, 8));
+   }
+
 }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantApi.java

@@ -17,11 +17,14 @@
 package com.kingsrook.qbits.assistant.api;
 
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.kingsrook.qbits.assistant.AssistantQBitConfig;
@@ -45,17 +48,20 @@ import com.kingsrook.qbits.assistant.model.AssistantToolCall;
 import com.kingsrook.qbits.assistant.model.AssistantTurn;
 import com.kingsrook.qbits.assistant.model.AssistantTurnEvent;
 import com.kingsrook.qbits.assistant.persistence.AssistantStore;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.tools.UiApiToolbox;
 import com.kingsrook.qbits.assistant.tools.bulkload.BulkLoadTools;
 import com.kingsrook.qqq.backend.core.actions.ai.AiToolOutcome;
 import com.kingsrook.qqq.backend.core.actions.ai.AiToolResult;
 import com.kingsrook.qqq.backend.core.actions.ai.output.AiRedactor;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.session.QSession;
 import com.kingsrook.qqq.backend.core.utils.JsonUtils;
 import com.kingsrook.qqq.backend.core.utils.StringUtils;
 import com.kingsrook.qqq.middleware.javalin.specs.invoke.UiApiCredentials;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
 
 /*******************************************************************************
@@ -65,6 +71,8 @@ import com.kingsrook.qqq.middleware.javalin.specs.invoke.UiApiCredentials;
  *******************************************************************************/
 public class AssistantApi
 {
+   private static final QLogger LOG = QLogger.getLogger(AssistantApi.class);
+
    public static final int TITLE_MAX_CHARS = 60;
 
    private final AssistantRuntime    runtime;
@@ -289,6 +297,19 @@ public class AssistantApi
    public RunningTurn startTurn(Integer conversationId, Map<String, Object> body) throws QException, AssistantApiException
    {
       requireAccess();
+      if(runtime.isShuttingDown())
+      {
+         throw (new AssistantApiException(503, "SHUTTING_DOWN", "The server is restarting. Try again in a moment."));
+      }
+      String concurrency = config.getDefaultConcurrencyMode();
+      if(body.containsKey("concurrency"))
+      {
+         if(!(body.get("concurrency") instanceof String mode) || !("supersede".equalsIgnoreCase(mode) || "concurrent".equalsIgnoreCase(mode)))
+         {
+            throw (new AssistantApiException(400, "concurrency must be supersede or concurrent."));
+         }
+         concurrency = mode.toUpperCase(Locale.ROOT);
+      }
       AssistantConversation conversation = requireConversation(conversationId);
 
       if(!(body.get("text") instanceof String text) || !StringUtils.hasContent(text))
@@ -324,30 +345,43 @@ public class AssistantApi
          throw (new AssistantApiException(429, "BUDGET", budget.message()));
       }
 
-      long running = 0;
-      for(AssistantTurn other : store.listRunningTurns(userId))
-      {
-         if(isRunning(reapIfOrphaned(other)))
-         {
-            running++;
-         }
-      }
-      if(running >= config.getMaxConcurrentTurnsPerUser())
-      {
-         throw (new AssistantApiException(429, "RATE_LIMIT", "You already have an assistant answer in progress. Wait for it to finish or stop it."));
-      }
-
       String limited = runtime.getRateLimiter().tryAcquire(userId, config.getMaxTurnsPerUserPerMinute(), config.getMaxConcurrentTurnsPerUser());
       if(limited != null)
       {
          throw (new AssistantApiException(429, "RATE_LIMIT", limited));
       }
 
+      List<AssistantTurn> remaining = new ArrayList<>();
+      for(AssistantTurn other : store.listRunningTurns(userId))
+      {
+         other = reapIfOrphaned(other);
+         if(!isRunning(other) || "SUPERSEDED".equals(other.getCancelReason()))
+         {
+            continue;
+         }
+         if("SUPERSEDE".equals(concurrency) && conversationId.equals(other.getConversationId()))
+         {
+            requestStop(other, "SUPERSEDED");
+         }
+         else
+         {
+            remaining.add(other);
+         }
+      }
+      remaining.sort(Comparator.comparing(AssistantTurn::getStartedAt, Comparator.nullsFirst(Comparator.naturalOrder()))
+         .thenComparing(AssistantTurn::getId));
+      while(remaining.size() >= config.getMaxConcurrentTurnsPerUser())
+      {
+         requestStop(remaining.removeFirst(), "SUPERSEDED");
+      }
+
+      TranscriptRedactor transcript = new TranscriptRedactor(runtime.getQInstance(), config.getClassificationSource());
+      transcript.attach();
       try
       {
          if(!StringUtils.hasContent(conversation.getTitle()))
          {
-            String title = deriveTitle(text);
+            String title = text;
             store.updateFields(AssistantConversation.TABLE_NAME, new QRecord().withValue("id", conversationId).withValue("title", title));
             conversation.setTitle(title);
          }
@@ -361,13 +395,18 @@ public class AssistantApi
             .withSeq(seq)
             .withPartsJson(new MessageParts().addText(text).toJson()));
 
+         Instant startedAt = runtime.getClock().instant();
          AssistantTurn turn = store.insert(new AssistantTurn()
             .withConversationId(conversationId)
             .withUserId(userId)
             .withUserMessageId(userMessage.getId())
             .withStatus("RUNNING")
             .withModel(active.model())
-            .withStartedAt(runtime.getClock().instant())
+            .withStartedAt(startedAt)
+            .withHeartbeatAt(startedAt)
+            .withLeaseOwner(runtime.getInstanceId())
+            .withPhase("starting")
+            .withConcurrencyMode(concurrency)
             .withToolCallCount(0)
             .withModelCallCount(0));
 
@@ -385,17 +424,49 @@ public class AssistantApi
          store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", turn.getId()).withValue("assistantMessageId", assistantMessage.getId()));
          turn.setAssistantMessageId(assistantMessage.getId());
 
-         RunningTurn runningTurn = runtime.register(new RunningTurn(turn.getId(), userId));
+         RunningTurn runningTurn = new RunningTurn(turn.getId(), userId);
          TurnEngine  engine      = new TurnEngine(runtime, session, userId, conversation, userMessage, assistantMessage, turn, text, pageContext, policy, runningTurn,
-            budget.softExceeded() ? budget.message() : null).withCredentials(credentials);
-         runtime.getExecutor().submit(engine);
+            budget.softExceeded() ? budget.message() : null).withCredentials(credentials).withTranscript(transcript);
+         try
+         {
+            if(config.getRunTracker() != null)
+            {
+               ///////////////////////////////////////////////////////////////////
+               // never the conversation title: it is derived from the prompt  //
+               ///////////////////////////////////////////////////////////////////
+               runningTurn.setTrackingId(config.getRunTracker().start(userId, turn.getId(), conversationId, "Assistant answer"));
+               config.getRunTracker().onCancel(runningTurn.getTrackingId(), () -> runningTurn.cancel("USER"));
+            }
+         }
+         catch(Exception e)
+         {
+            LOG.warn("Assistant run tracker start failed", logPair("turnId", turn.getId()));
+         }
+         runtime.register(runningTurn);
+         if(!runningTurn.isFinalized())
+         {
+            try
+            {
+               runtime.getExecutor().submit(engine);
+            }
+            catch(RuntimeException e)
+            {
+               runningTurn.forceFinalize("SHUTDOWN");
+               throw (e);
+            }
+         }
          return (runningTurn);
       }
       catch(QException | RuntimeException e)
       {
+         transcript.close();
          runtime.getRateLimiter().release(userId);
          throw (e);
       }
+      finally
+      {
+         TranscriptRedactor.detach();
+      }
    }
 
 
@@ -443,22 +514,94 @@ public class AssistantApi
    {
       requireAccess();
       AssistantTurn turn = requireTurn(turnId);
-      if(runtime.getRunningTurn(turnId) == null)
+      if(isRunning(turn))
+      {
+         turn = requestStop(turn, "USER");
+      }
+      return (Map.of("status", turn.getStatus()));
+   }
+
+
+
+   /*******************************************************************************
+    ** Durable Stop, shared by the API and superseding another question.
+    *******************************************************************************/
+   private AssistantTurn requestStop(AssistantTurn turn, String reason) throws QException
+   {
+      synchronized(store)
+      {
+         turn = store.getTurn(userId, turn.getId());
+         if(!isRunning(turn))
+         {
+            return (turn);
+         }
+         RunningTurn live = runtime.getRunningTurn(turn.getId());
+         if(live == null && orphaned(turn))
+         {
+            return (reap(turn, reason));
+         }
+         Instant now = runtime.getClock().instant();
+         if(!store.updateRunningTurn(userId, new QRecord().withValue("id", turn.getId())
+            .withValue("status", "CANCELLING").withValue("cancelReason", reason).withValue("cancelRequestedAt", now)))
+         {
+            return (store.getTurn(userId, turn.getId()));
+         }
+         turn.setStatus("CANCELLING");
+         turn.setCancelReason(reason);
+         turn.setCancelRequestedAt(now);
+         if(live != null)
+         {
+            try
+            {
+               live.cancel(reason);
+            }
+            catch(Exception e)
+            {
+               LOG.warn("Assistant stream cancellation failed; watchdog will finalize");
+            }
+         }
+         return (turn);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** GET /turns/active: the caller's runs, newest first, after orphan recovery.
+    *******************************************************************************/
+   public Map<String, Object> activeTurns() throws QException, AssistantApiException
+   {
+      requireAccess();
+      List<AssistantTurn> turns = new ArrayList<>();
+      for(AssistantTurn turn : store.listRunningTurns(userId))
       {
          turn = reapIfOrphaned(turn);
+         if(isRunning(turn))
+         {
+            turns.add(turn);
+         }
       }
-      if(!isRunning(turn))
+      turns.sort(Comparator.comparing(AssistantTurn::getStartedAt, Comparator.nullsFirst(Comparator.naturalOrder()))
+         .thenComparing(AssistantTurn::getId).reversed());
+      List<Map<String, Object>> rows = new ArrayList<>();
+      for(AssistantTurn turn : turns)
       {
-         throw (new AssistantApiException(409, "This answer has already finished."));
+         RunningTurn live = runtime.getRunningTurn(turn.getId());
+         Map<String, Object> row = new LinkedHashMap<>();
+         row.put("turnId", turn.getId());
+         row.put("conversationId", turn.getConversationId());
+         row.put("assistantMessageId", turn.getAssistantMessageId());
+         row.put("userMessageId", turn.getUserMessageId());
+         row.put("status", turn.getStatus());
+         row.put("phase", live == null ? turn.getPhase() : live.getPhase());
+         row.put("startedAt", turn.getStartedAt() == null ? null : turn.getStartedAt().toString());
+         row.put("heartbeatAt", turn.getHeartbeatAt() == null ? null : turn.getHeartbeatAt().toString());
+         row.put("elapsedMs", turn.getStartedAt() == null ? 0 : Math.max(0, Duration.between(turn.getStartedAt(), runtime.getClock().instant()).toMillis()));
+         row.put("live", live != null);
+         row.put("concurrencyMode", turn.getConcurrencyMode());
+         rows.add(row);
       }
-
-      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", turnId).withValue("status", "CANCELLING"));
-      RunningTurn live = runtime.getRunningTurn(turnId);
-      if(live != null && userId.equals(live.getUserId()))
-      {
-         live.cancel("USER");
-      }
-      return (Map.of("status", "CANCELLING"));
+      return (Map.of("turns", rows));
    }
 
 
@@ -474,60 +617,106 @@ public class AssistantApi
 
 
    /*******************************************************************************
-    ** A turn whose replica died: it still says RUNNING / CANCELLING, no turn
-    ** of this runtime owns it, and it started longer ago than maxTurnSeconds
-    ** + 60 s (a live engine always ends itself by then). On first sight it is
-    ** finished as ERROR: the assistant message gets status ERROR and a
-    ** notice, and synthetic error (INTERNAL) + message.end events are
-    ** persisted, so replays end. Returns the (possibly updated) turn.
+    ** A remote run loses its lease, exceeds the absolute deadline, or has a
+    ** cancellation request that its owner did not acknowledge in time.
+    *******************************************************************************/
+   private boolean orphaned(AssistantTurn turn)
+   {
+      Instant now = runtime.getClock().instant();
+      Instant heartbeat = turn.getHeartbeatAt() == null ? turn.getStartedAt() : turn.getHeartbeatAt();
+      return ((heartbeat != null && heartbeat.isBefore(now.minusSeconds(config.getTurnLeaseSeconds())))
+         || (turn.getStartedAt() != null && turn.getStartedAt().isBefore(now.minusSeconds(maxStreamSeconds())))
+         || ("CANCELLING".equals(turn.getStatus()) && turn.getCancelRequestedAt() != null
+         && turn.getCancelRequestedAt().isBefore(now.minusSeconds(config.getTurnLeaseSeconds()))));
+   }
+
+
+
+   /*******************************************************************************
+    ** Recover an abandoned run and leave a terminal replay for refreshes.
     *******************************************************************************/
    AssistantTurn reapIfOrphaned(AssistantTurn turn) throws QException
    {
-      if(!isRunning(turn) || runtime.getRunningTurn(turn.getId()) != null || turn.getStartedAt() == null)
+      synchronized(store)
       {
-         return (turn);
-      }
-      Instant stale = runtime.getClock().instant().minusSeconds(config.getMaxTurnSeconds() + 60L);
-      if(!turn.getStartedAt().isBefore(stale))
-      {
-         return (turn);
-      }
-
-      String text = "This answer was interrupted (the server restarted). Ask again.";
-      if(turn.getAssistantMessageId() != null)
-      {
-         AssistantMessage message = store.getMessage(userId, turn.getAssistantMessageId());
-         if(message != null)
+         if(!isRunning(turn) || runtime.getRunningTurn(turn.getId()) != null || !orphaned(turn))
          {
-            store.updateFields(AssistantMessage.TABLE_NAME, new QRecord()
-               .withValue("id", message.getId())
-               .withValue("status", "ERROR")
-               .withValue("partsJson", MessageParts.fromJson(message.getPartsJson()).addNotice("error", text).toJson()));
+            return (turn);
+         }
+         String reason = "CANCELLING".equals(turn.getStatus()) && turn.getCancelRequestedAt() != null
+            && turn.getCancelRequestedAt().isBefore(runtime.getClock().instant().minusSeconds(config.getTurnLeaseSeconds()))
+            ? (turn.getCancelReason() == null ? "USER" : turn.getCancelReason()) : null;
+         return (reap(turn, reason));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Re-read before recovery; never overwrite an already terminal row or
+    ** append another terminal event. Each persistence step is independent.
+    *******************************************************************************/
+   private AssistantTurn reap(AssistantTurn turn, String cancelReason) throws QException
+   {
+      AssistantTurn current = store.getTurn(userId, turn.getId());
+      if(!isRunning(current))
+      {
+         return (current);
+      }
+      if(!orphaned(current))
+      {
+         return (cancelReason == null ? current : requestStop(current, cancelReason));
+      }
+      // Recovery only appends fixed notices to already redacted stored parts (or
+      // crash placeholders). It never reloads model/tool values or raw user text.
+      String status = cancelReason == null ? "ERROR" : "CANCELLED";
+      String text = cancelReason == null ? "This answer was interrupted (the server restarted). Ask again."
+         : ("SUPERSEDED".equals(cancelReason) ? TurnEngine.SUPERSEDED_TEXT : "Stopped by you.");
+      if(current.getAssistantMessageId() != null)
+      {
+         try
+         {
+            AssistantMessage message = store.getMessage(userId, current.getAssistantMessageId());
+            if(message != null)
+            {
+               store.updateFields(AssistantMessage.TABLE_NAME, new QRecord().withValue("id", message.getId()).withValue("status", status)
+                  .withValue("partsJson", MessageParts.fromJson(message.getPartsJson()).addNotice(cancelReason == null ? "error" : "info", text).toJson()));
+            }
+         }
+         catch(Exception e)
+         {
+            LOG.warn("Could not persist recovered assistant message");
          }
       }
-
-      int seq = store.lastEventSeq(userId, turn.getId());
-      store.insertEvent(turn.getId(), userId, seq + 1, TurnEvent.ERROR, new TurnEvent(seq + 1, TurnEvent.ERROR, Map.of("code", "INTERNAL", "message", text)).toJson());
-      Map<String, Object> end = new java.util.LinkedHashMap<>();
-      if(turn.getAssistantMessageId() != null)
+      try
       {
-         end.put("messageId", turn.getAssistantMessageId());
+         int seq = store.lastEventSeq(userId, current.getId());
+         List<AssistantTurnEvent> last = store.listEvents(userId, current.getId(), Math.max(0, seq - 1), 1);
+         if(last.isEmpty() || !TurnEvent.MESSAGE_END.equals(last.getFirst().getEventType()))
+         {
+            if(cancelReason == null)
+            {
+               seq++;
+               store.insertEvent(current.getId(), userId, seq, TurnEvent.ERROR,
+                  new TurnEvent(seq, TurnEvent.ERROR, Map.of("code", "INTERNAL", "message", text)).toJson());
+            }
+            seq++;
+            Map<String, Object> end = new LinkedHashMap<>();
+            end.put("messageId", current.getAssistantMessageId());
+            end.put("status", status);
+            end.put("stopReason", cancelReason == null ? "interrupted" : "cancelled");
+            store.insertEvent(current.getId(), userId, seq, TurnEvent.MESSAGE_END, new TurnEvent(seq, TurnEvent.MESSAGE_END, end).toJson());
+         }
+      }
+      catch(Exception e)
+      {
+         LOG.warn("Could not persist recovered assistant events");
       }
-      end.put("status", "ERROR");
-      end.put("stopReason", "interrupted");
-      store.insertEvent(turn.getId(), userId, seq + 2, TurnEvent.MESSAGE_END, new TurnEvent(seq + 2, TurnEvent.MESSAGE_END, end).toJson());
-
       Instant now = runtime.getClock().instant();
-      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord()
-         .withValue("id", turn.getId())
-         .withValue("status", "ERROR")
-         .withValue("errorCode", "INTERNAL")
-         .withValue("stopReason", "interrupted")
-         .withValue("endedAt", now));
-      turn.setStatus("ERROR");
-      turn.setErrorCode("INTERNAL");
-      turn.setEndedAt(now);
-      return (turn);
+      store.updateRunningTurn(userId, new QRecord().withValue("id", current.getId()).withValue("status", status)
+         .withValue("errorCode", cancelReason == null ? "INTERNAL" : null).withValue("cancelReason", cancelReason)
+         .withValue("stopReason", cancelReason == null ? "interrupted" : "cancelled").withValue("phase", "done").withValue("endedAt", now));
+      return (store.getTurn(userId, current.getId()));
    }
 
 
@@ -591,6 +780,12 @@ public class AssistantApi
          throw (new AssistantApiException(409, "This action is already " + action.getStatus().toLowerCase() + "."));
       }
 
+      AssistantTurn proposingTurn = store.getTurn(userId, action.getTurnId());
+      if(proposingTurn != null && isRunning(proposingTurn))
+      {
+         throw (new AssistantApiException(409, "Wait for the assistant to finish before confirming this action."));
+      }
+
       ///////////////////////////////////////////////////////////////////////
       // claim it first, so a double click cannot run the write twice     //
       ///////////////////////////////////////////////////////////////////////
@@ -786,6 +981,16 @@ public class AssistantApi
 
 
 
+   /*******************************************************************************
+    ** Hard upper bound for remote stream polling and stale-run recovery.
+    *******************************************************************************/
+   public long maxStreamSeconds()
+   {
+      return (config.getMaxTurnSeconds() + (long) config.getTurnForceFinalizeGraceSeconds() + config.getTurnLeaseSeconds());
+   }
+
+
+
    /*******************************************************************************
     ** The user may use the panel: enabled, signed in, assistant.hasAccess.
     *******************************************************************************/

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/AssistantRoutes.java

@@ -17,12 +17,12 @@
 package com.kingsrook.qbits.assistant.api;
 
 
+import com.kingsrook.qbits.assistant.api.specs.AssistantSpecs;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
-import com.kingsrook.qbits.assistant.api.specs.AssistantSpecs;
 import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
-import com.kingsrook.qqq.middleware.javalin.specs.extensions.MiddlewareSpecExtensionsMetaData;
 import com.kingsrook.qqq.middleware.javalin.QJavalinMetaData;
+import com.kingsrook.qqq.middleware.javalin.specs.extensions.MiddlewareSpecExtensionsMetaData;
 
 
 /*******************************************************************************

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/TurnStreams.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
+import java.util.Map;
 import com.kingsrook.qbits.assistant.engine.TurnEvent;
 import com.kingsrook.qbits.assistant.engine.TurnEventLog;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
@@ -101,7 +102,7 @@ public final class TurnStreams
          OutputStream out       = startStream(context);
          int          last      = after;
          long         lastWrite = System.currentTimeMillis();
-         long         giveUpAt  = System.currentTimeMillis() + (api.maxTurnSeconds() + 60L) * 1000L;
+         long         giveUpAt  = System.currentTimeMillis() + api.maxStreamSeconds() * 1000L;
          AssistantApi.TurnEvents current = first;
          while(true)
          {
@@ -113,10 +114,16 @@ public final class TurnStreams
                lastWrite = System.currentTimeMillis();
                ended |= event.isTerminal();
             }
-            if(ended || (current.finished() && !current.more()) || System.currentTimeMillis() > giveUpAt)
+            if(ended || (current.finished() && !current.more()))
             {
                return;
             }
+            if(System.currentTimeMillis() > giveUpAt)
+            {
+               write(out, new TurnEvent(last + 1, TurnEvent.ERROR, Map.of("code", "TIMEOUT", "message", "The event stream timed out. Reconnect to check this answer.")).toSseFrame());
+               write(out, new TurnEvent(last + 2, TurnEvent.MESSAGE_END, Map.of("status", "ERROR", "stopReason", "stream_timeout")).toSseFrame());
+               return;
+            }
             if(!current.more())
             {
                Thread.sleep(REMOTE_POLL_MS);
@@ -153,7 +160,7 @@ public final class TurnStreams
    /*******************************************************************************
     **
     *******************************************************************************/
-   private static void writeLive(OutputStream out, TurnEventLog log, int after) throws IOException, InterruptedException
+   static void writeLive(OutputStream out, TurnEventLog log, int after) throws IOException, InterruptedException
    {
       int last = after;
       while(true)
@@ -163,7 +170,12 @@ public final class TurnStreams
          {
             if(log.isClosed())
             {
-               return;
+               // The terminal event may have arrived after the empty poll.
+               if(log.size() <= last)
+               {
+                  return;
+               }
+               continue;
             }
             write(out, ping());
             continue;

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/api/specs/AssistantSpecs.java

@@ -61,7 +61,7 @@ import io.javalin.http.HttpStatus;
 public final class AssistantSpecs
 {
    public static final String EXTENSION_NAME    = "assistant";
-   public static final String EXTENSION_VERSION = "1.1.0";
+   public static final String EXTENSION_VERSION = "1.2.0";
    public static final String PREFIX            = "/assistant";
 
 
@@ -94,6 +94,7 @@ public final class AssistantSpecs
          .withSpec(StartTurnSpec.class)
          .withSpec(TurnEventsSpec.class)
          .withSpec(CancelTurnSpec.class)
+         .withSpec(ActiveTurnsSpec.class)
          .withSpec(FeedbackSpec.class)
          .withSpec(GetActionSpec.class)
          .withSpec(ConfirmActionSpec.class)
@@ -218,9 +219,14 @@ public final class AssistantSpecs
          "href", string("After a confirm: the written record's /app link.").withNullable(true)));
       rs.put("AssistantAccessResponseV1", object("Explain access: per table, the user's UI permissions, the assistant's data access, and each tool's verdict.",
          "tables", array(freeObject("table, label, dataAccess, dataAccessSource, ui {visible, read, insert, edit, delete}, operations [{tool, operation, allowed, reason}]."), null)));
-      rs.put("AssistantTurnEventV1", object("One server-sent event of a turn (event: <type>, id: <seq>, data: this object). Types: message.start, text.delta, tool.start, tool.result, ui.navigate, notice, usage, message.end, error, ping.",
+      rs.put("AssistantActiveTurnsResponseV1", object("The caller's active runs, newest first; stale leases are reaped.",
+         "turns", array(object("An active assistant run.", "turnId", integer(null), "conversationId", integer(null),
+            "assistantMessageId", integer(null), "userMessageId", integer(null), "status", string("RUNNING or CANCELLING."),
+            "phase", string(null), "startedAt", string("ISO-8601."), "heartbeatAt", string("ISO-8601."),
+            "elapsedMs", integer(null), "live", bool("Running on this replica."), "concurrencyMode", string("SUPERSEDE or CONCURRENT.")), null)));
+      rs.put("AssistantTurnEventV1", object("One server-sent event of a turn (event: <type>, id: <seq>, data: this object). Types: message.start, turn.phase, text.delta, tool.start, tool.result, ui.navigate, notice, usage, message.end, error, ping.",
          "seq", integer("Sequence number; resume with ?after=<seq>."), "type", string("The event type (also the SSE event name)."),
-         "data", freeObject("The event's fields, by type.")));
+         "data", freeObject("The event's fields, by type. message.start includes startedAt (ISO-8601). turn.phase includes phase (thinking, tool, writing), optional toolName and label.")));
       rs.put("AssistantErrorV1", object("An error.", "error", string("A message for the user."), "code", string("BUDGET, RATE_LIMIT, NOT_CONFIGURED, KILL_SWITCH, DISABLED, NO_PERMISSION...").withNullable(true)));
       return (rs);
    }
@@ -562,6 +568,67 @@ public final class AssistantSpecs
 
 
 
+   /*******************************************************************************
+    ** GET /assistant/turns/active
+    *******************************************************************************/
+   public static class ActiveTurnsSpec extends AssistantJsonSpec
+   {
+      /*******************************************************************************
+       ** HTTP method.
+       *******************************************************************************/
+      @Override
+      protected HttpMethod method()
+      {
+         return (HttpMethod.GET);
+      }
+
+
+
+      /*******************************************************************************
+       ** Route path.
+       *******************************************************************************/
+      @Override
+      protected String path()
+      {
+         return (PREFIX + "/turns/active");
+      }
+
+
+
+      /*******************************************************************************
+       ** Operation summary.
+       *******************************************************************************/
+      @Override
+      protected String summary()
+      {
+         return ("Find the user's active assistant runs");
+      }
+
+
+
+      /*******************************************************************************
+       ** Response schema.
+       *******************************************************************************/
+      @Override
+      protected String getResponseSchemaName()
+      {
+         return ("AssistantActiveTurnsResponseV1");
+      }
+
+
+
+      /*******************************************************************************
+       ** Runs in the invoker's authenticated user session.
+       *******************************************************************************/
+      @Override
+      protected Object handle(AssistantApi api, JsonInput input) throws Exception
+      {
+         return (api.activeTurns());
+      }
+   }
+
+
+
    /*******************************************************************************
     ** POST /assistant/conversations
     *******************************************************************************/
@@ -844,7 +911,7 @@ public final class AssistantSpecs
       @Override
       protected Schema defineRequestBodySchema()
       {
-         return (object("The user's text and the page context hint.", "text", string("The user's message."), "pageContext", freeObject("Where the user is in the app.")));
+         return (object("The user's text and the page context hint.", "text", string("The user's message."), "pageContext", freeObject("Where the user is in the app."), "concurrency", string("supersede or concurrent (case-insensitive); defaults to the host configuration.")));
       }
 
 
@@ -1038,10 +1105,24 @@ public final class AssistantSpecs
 
 
 
+      /*******************************************************************************
+       ** Older replicas can still return 202 during a rolling upgrade. New
+       ** replicas return 200 for both active and already finished turns.
+       *******************************************************************************/
+      @Override
+      public List<BasicResponse> defineAdditionalBasicResponses()
+      {
+         List<BasicResponse> responses = new java.util.ArrayList<>(super.defineAdditionalBasicResponses());
+         responses.add(new BasicResponse(HttpStatus.ACCEPTED, "Legacy replica: cancellation accepted", "AssistantStatusResponseV1"));
+         return (responses);
+      }
+
+
+
       @Override
       protected HttpStatus successStatus()
       {
-         return (HttpStatus.ACCEPTED);
+         return (HttpStatus.OK);
       }
    }
 

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/classification/ConfiguredClassificationSource.java

@@ -80,7 +80,7 @@ public class ConfiguredClassificationSource implements ClassificationSource
    @Override
    public DataClassification ofTable(String table)
    {
-      return (table == null || table.startsWith("assistant_") ? DataClassification.PUBLIC : Classifications.max(null, tables.get(table)));
+      return (table == null || table.startsWith("assistant_") ? DataClassification.PUBLIC : Classifications.max(null, tables.get(table) == null ? AiDataClassifications.ofTable(coreTable(table)) : tables.get(table)));
    }
 
 

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RateLimiter.java

@@ -26,15 +26,12 @@ import java.util.Map;
 
 
 /*******************************************************************************
- ** Per-user turn rate limits (spec 3.9): turns started per minute and turns
- ** running at once. In memory, per replica - with several replicas the
- ** effective limit is per replica (the concurrent limit is also checked
- ** against the turn table by the API).
+ ** Per-user starts per minute, in memory per replica. The API implements
+ ** concurrency by superseding turns, never by refusing another question.
  *******************************************************************************/
 public class RateLimiter
 {
    private final Map<String, Deque<Instant>> starts  = new HashMap<>();
-   private final Map<String, Integer>        running = new HashMap<>();
    private final Clock                       clock;
 
 
@@ -61,17 +58,12 @@ public class RateLimiter
          queue.pollFirst();
       }
 
-      if(running.getOrDefault(userId, 0) >= concurrent)
-      {
-         return ("You already have an assistant answer in progress. Wait for it to finish or stop it.");
-      }
       if(queue.size() >= perMinute)
       {
          return ("Too many assistant requests in the last minute. Try again shortly.");
       }
 
       queue.addLast(now);
-      running.merge(userId, 1, Integer::sum);
       return (null);
    }
 
@@ -82,7 +74,7 @@ public class RateLimiter
     *******************************************************************************/
    public synchronized void release(String userId)
    {
-      running.computeIfPresent(userId, (k, v) -> v <= 1 ? null : v - 1);
+      // Only starts per minute are counted; completion does not refund a start.
    }
 
 }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/RunningTurn.java

@@ -17,7 +17,10 @@
 package com.kingsrook.qbits.assistant.engine;
 
 
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 import com.kingsrook.qbits.assistant.spi.StreamHandle;
 
 
@@ -30,6 +33,14 @@ public class RunningTurn
    private final String                        userId;
    private final TurnEventLog                  log          = new TurnEventLog();
    private final AtomicReference<String>       cancelReason = new AtomicReference<>();
+   private final AtomicBoolean finalized = new AtomicBoolean();
+   private Instant startedAt;
+   private volatile String phase = "starting";
+   private volatile String toolName;
+   private volatile Thread engineThread;
+   private volatile Consumer<String> forceFinalizer;
+   private volatile long cancelledNanos;
+   private volatile String trackingId;
    private volatile StreamHandle               currentHandle;
 
 
@@ -39,6 +50,7 @@ public class RunningTurn
     *******************************************************************************/
    public RunningTurn(Integer turnId, String userId)
    {
+      this.startedAt = Instant.now();
       this.turnId = turnId;
       this.userId = userId;
    }
@@ -50,12 +62,24 @@ public class RunningTurn
     *******************************************************************************/
    public void cancel(String reason)
    {
-      if(cancelReason.compareAndSet(null, reason))
+      if(!finalized.get() && cancelReason.compareAndSet(null, reason))
       {
-         StreamHandle handle = currentHandle;
-         if(handle != null)
+         cancelledNanos = System.nanoTime();
+         try
          {
-            handle.cancel();
+            StreamHandle handle = currentHandle;
+            if(handle != null)
+            {
+               handle.cancel();
+            }
+         }
+         finally
+         {
+            Thread thread = engineThread;
+            if(thread != null && thread != Thread.currentThread())
+            {
+               thread.interrupt();
+            }
          }
       }
    }
@@ -124,4 +148,140 @@ public class RunningTurn
       return (log);
    }
 
+
+
+
+   /*******************************************************************************
+    ** Claim terminal processing exactly once.
+    *******************************************************************************/
+   public boolean tryFinalize()
+   {
+      return (finalized.compareAndSet(false, true));
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether terminal processing has been claimed.
+    *******************************************************************************/
+   public boolean isFinalized()
+   {
+      return (finalized.get());
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for startedAt
+    *******************************************************************************/
+   public Instant getStartedAt()
+   {
+      return (startedAt);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for startedAt
+    *******************************************************************************/
+   public void setStartedAt(Instant startedAt)
+   {
+      this.startedAt = startedAt;
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for phase
+    *******************************************************************************/
+   public String getPhase()
+   {
+      return (phase);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for toolName
+    *******************************************************************************/
+   public String getToolName()
+   {
+      return (toolName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Record the current phase and optional tool.
+    *******************************************************************************/
+   public void setPhase(String phase, String toolName)
+   {
+      this.phase = phase;
+      this.toolName = toolName;
+   }
+
+
+
+   /*******************************************************************************
+    ** Record the thread to interrupt when cancelling.
+    *******************************************************************************/
+   public void setEngineThread(Thread engineThread)
+   {
+      this.engineThread = engineThread;
+   }
+
+
+
+   /*******************************************************************************
+    ** Install the engine finalizer before registration.
+    *******************************************************************************/
+   public void setForceFinalizer(Consumer<String> forceFinalizer)
+   {
+      this.forceFinalizer = forceFinalizer;
+   }
+
+
+
+   /*******************************************************************************
+    ** Force terminal processing using the original user session.
+    *******************************************************************************/
+   public void forceFinalize(String reason)
+   {
+      Consumer<String> finalizer = forceFinalizer;
+      if(finalizer != null)
+      {
+         finalizer.accept(reason);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether cancellation has exceeded its bounded grace period.
+    *******************************************************************************/
+   public boolean cancelGraceElapsed(int seconds)
+   {
+      return (isCancelled() && cancelledNanos != 0 && System.nanoTime() - cancelledNanos >= seconds * 1_000_000_000L);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for trackingId
+    *******************************************************************************/
+   public String getTrackingId()
+   {
+      return (trackingId);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for trackingId
+    *******************************************************************************/
+   public void setTrackingId(String trackingId)
+   {
+      this.trackingId = trackingId;
+   }
+
 }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEngine.java

@@ -44,6 +44,7 @@ import com.kingsrook.qbits.assistant.model.AssistantScheduledTask;
 import com.kingsrook.qbits.assistant.model.AssistantToolCall;
 import com.kingsrook.qbits.assistant.model.AssistantTurn;
 import com.kingsrook.qbits.assistant.persistence.AssistantStore;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.scheduled.ScheduledTurn;
 import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
 import com.kingsrook.qbits.assistant.spi.CacheHint;
@@ -102,10 +103,11 @@ public class TurnEngine implements Runnable
 {
    private static final QLogger LOG = QLogger.getLogger(TurnEngine.class);
 
-   private static final int MAX_INPUT_JSON_CHARS  = 4000;
+   public static final String SUPERSEDED_TEXT = "Stopped because you asked something else.";
+   public static final String TIMEOUT_TEXT = "The assistant took too long to answer, so it was stopped. Ask again.";
+   public static final String RESTART_TEXT = "The server restarted while answering. Ask again.";
 
    private static final AiRedactor INPUT_REDACTOR = new AiRedactor(List.of());
-   private static final int MAX_RESULT_JSON_CHARS = 64_000;
    private static final int TEXT_FLUSH_CHARS      = 48;
    private static final long TEXT_FLUSH_MILLIS    = 120;
 
@@ -151,6 +153,8 @@ public class TurnEngine implements Runnable
 
    private volatile boolean finished = false;
    private UiApiCredentials credentials;
+   private volatile TranscriptRedactor transcript;
+   private boolean engineRunning;
 
 
 
@@ -175,6 +179,10 @@ public class TurnEngine implements Runnable
       this.policy = policy;
       this.running = running;
       this.softBudgetNotice = softBudgetNotice;
+      this.startedAt = turn.getStartedAt() == null ? runtime.getClock().instant() : turn.getStartedAt();
+      this.deadline = startedAt.plusSeconds(config.getMaxTurnSeconds());
+      running.setStartedAt(startedAt);
+      running.setForceFinalizer(this::forceFinish);
    }
 
 
@@ -192,14 +200,43 @@ public class TurnEngine implements Runnable
 
 
 
+   /*******************************************************************************
+    ** Continue the API's provisional transcript scope on the engine thread.
+    *******************************************************************************/
+   public TurnEngine withTranscript(TranscriptRedactor transcript)
+   {
+      this.transcript = transcript;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Finalization may run before submission or on a watchdog/shutdown thread.
+    ** Bind the same turn scope before any of those paths writes transcript content.
+    *******************************************************************************/
+   private synchronized void attachTranscript()
+   {
+      if(transcript == null)
+      {
+         transcript = new TranscriptRedactor(runtime.getQInstance(), config.getClassificationSource());
+      }
+      transcript.attach();
+   }
+
+
+
    /*******************************************************************************
     ** Run the turn to completion. Never throws.
     *******************************************************************************/
    @Override
    public void run()
    {
-      startedAt = runtime.getClock().instant();
-      deadline = startedAt.plusSeconds(config.getMaxTurnSeconds());
+      if(running.isFinalized())
+      {
+         return;
+      }
+      running.setEngineThread(Thread.currentThread());
 
       String status     = "COMPLETE";
       String stopReason = null;
@@ -209,6 +246,16 @@ public class TurnEngine implements Runnable
       try
       {
          QContext.init(runtime.getQInstance(), session);
+         synchronized(this)
+         {
+            if(running.isFinalized())
+            {
+               return;
+            }
+            engineRunning = true;
+            attachTranscript();
+         }
+         startWatchdog();
          scheduledTask = runtime.getScheduledTurn().task(conversation.getId());
          if(conversation.getScheduledTaskId() != null && scheduledTask == null)
          {
@@ -218,9 +265,9 @@ public class TurnEngine implements Runnable
          ScheduledTurn.checkCost(runtime, conversation.getId(), turn.getId());
          classification = new TurnClassification(config.getClassificationSource(), conversation, pageContext);
          store.saveClassification(userId, conversation.getId(), turn.getId(), classification);
-         startKillSwitchWatcher();
 
-         emit(TurnEvent.MESSAGE_START, map("turnId", turn.getId(), "conversationId", conversation.getId(), "userMessageId", userMessage.getId(), "messageId", assistantMessage.getId()));
+
+         emit(TurnEvent.MESSAGE_START, map("turnId", turn.getId(), "conversationId", conversation.getId(), "userMessageId", userMessage.getId(), "messageId", assistantMessage.getId(), "startedAt", startedAt.toString()));
          if(softBudgetNotice != null)
          {
             notice("warning", softBudgetNotice);
@@ -232,6 +279,7 @@ public class TurnEngine implements Runnable
          ModelRouter.Route route;
          try
          {
+            phase("thinking", null);
             route = router.route(session, conversation, turn, userText, pageContext, policy, classification, running, deadline, text -> notice("warning", text));
          }
          finally
@@ -420,44 +468,88 @@ public class TurnEngine implements Runnable
       }
       finally
       {
-         finished = true;
-         finish(status, stopReason, errorCode, errorText);
-         QContext.clear();
+         // Clear interruption while persisting the terminal state.
+         boolean interrupted = Thread.interrupted();
+         if(running.isCancelled())
+         {
+            Stop stop = cancelledStop();
+            status = stop.status();
+            stopReason = stop.stopReason();
+            errorCode = stop.errorCode();
+            errorText = stop.message();
+         }
+         try
+         {
+            finish(status, stopReason, errorCode, errorText);
+         }
+         finally
+         {
+            if(transcript != null)
+            {
+               transcript.close();
+            }
+            if(interrupted)
+            {
+               Thread.currentThread().interrupt();
+            }
+            QContext.clear();
+         }
       }
    }
 
 
 
    /*******************************************************************************
-    ** Poll the kill switch while the turn runs (with the user's own session),
-    ** so a flip cancels the turn - and every other turn here - even while it
-    ** waits on the model.
+    ** Renew the lease, poll cancellation and the kill switch with the user's
+    ** own session, and enforce a hard deadline even if a provider is wedged.
     *******************************************************************************/
-   private void startKillSwitchWatcher()
+   private void startWatchdog()
    {
-      long pollMillis = config.getKillSwitchPollMillis() == null ? 2000 : config.getKillSwitchPollMillis();
-      Thread.ofVirtual().name("assistant-killswitch-" + turn.getId()).start(() ->
+      long heartbeatMillis = config.getTurnHeartbeatSeconds() * 1000L;
+      long pollMillis = Math.min(heartbeatMillis, config.getKillSwitchPollMillis() == null ? 2000 : config.getKillSwitchPollMillis());
+      Thread.ofVirtual().name("assistant-watchdog-" + turn.getId()).start(() ->
       {
+         long lastHeartbeat = System.nanoTime();
          try
          {
-            while(!finished && !running.isCancelled())
+            QContext.init(runtime.getQInstance(), session);
+            while(!running.isFinalized())
             {
                Thread.sleep(pollMillis);
-               if(finished)
+               if(running.isFinalized())
                {
-                  return;
+                  break;
                }
-               QContext.init(runtime.getQInstance(), session);
                try
                {
+                  if(System.nanoTime() - lastHeartbeat >= heartbeatMillis * 1_000_000L)
+                  {
+                     lastHeartbeat = System.nanoTime();
+                     heartbeat();
+                  }
                   if(runtime.getPolicyService().isKillSwitchOn())
                   {
                      runtime.cancelAll("KILL_SWITCH");
                   }
                }
+               catch(Exception e)
+               {
+                  LOG.warn("Assistant watchdog tick failed", hipaaSafe(e), logPair("turnId", turn.getId()));
+               }
                finally
                {
-                  QContext.clear();
+                  // Deadlines do not depend on successful database or policy reads.
+                  Instant now = runtime.getClock().instant();
+                  if(!now.isBefore(deadline))
+                  {
+                     safely(() -> running.cancel("TIMEOUT"));
+                  }
+                  if(!now.isBefore(deadline.plusSeconds(config.getTurnForceFinalizeGraceSeconds()))
+                     || running.cancelGraceElapsed(config.getTurnForceFinalizeGraceSeconds()))
+                  {
+                     Stop stop = running.isCancelled() ? cancelledStop() : new Stop("ERROR", "TIMEOUT", TIMEOUT_TEXT, "timeout");
+                     finish(stop.status(), stop.stopReason(), stop.errorCode(), stop.message());
+                  }
                }
             }
          }
@@ -465,11 +557,63 @@ public class TurnEngine implements Runnable
          {
             Thread.currentThread().interrupt();
          }
+         finally
+         {
+            QContext.clear();
+         }
       });
    }
 
 
 
+   /*******************************************************************************
+    ** Renew only an active row, then observe persisted cancellation requests.
+    *******************************************************************************/
+   private synchronized void heartbeat() throws QException
+   {
+      if(running.isFinalized())
+      {
+         return;
+      }
+      AssistantTurn row = store.getTurn(userId, turn.getId());
+      if(row == null || !("RUNNING".equals(row.getStatus()) || "CANCELLING".equals(row.getStatus())))
+      {
+         safely(() -> running.cancel("USER"));
+         finish(row == null ? "ERROR" : row.getStatus(), row == null ? "interrupted" : row.getStopReason(),
+            row == null ? "INTERNAL" : row.getErrorCode(), null);
+         return;
+      }
+      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", turn.getId())
+         .withValue("heartbeatAt", runtime.getClock().instant()).withValue("phase", running.getPhase()));
+      row = store.getTurn(userId, turn.getId());
+      if(row != null && (row.getCancelRequestedAt() != null || "CANCELLING".equals(row.getStatus())))
+      {
+         running.cancel(row.getCancelReason() == null ? "USER" : row.getCancelReason());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Finalize from shutdown using the original authenticated user's context.
+    *******************************************************************************/
+   private void forceFinish(String reason)
+   {
+      try
+      {
+         QContext.init(runtime.getQInstance(), session);
+         safely(() -> running.cancel(reason));
+         Stop stop = "SHUTDOWN".equals(reason) ? new Stop("ERROR", "INTERRUPTED", RESTART_TEXT, "interrupted") : cancelledStop();
+         finish(stop.status(), stop.stopReason(), stop.errorCode(), stop.message());
+      }
+      finally
+      {
+         QContext.clear();
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Why a turn stops.
     *******************************************************************************/
@@ -504,14 +648,15 @@ public class TurnEngine implements Runnable
       }
       if(!runtime.getClock().instant().isBefore(deadline))
       {
-         return (new Stop("ERROR", "TIMEOUT", "The assistant ran out of time for this question.", "timeout"));
+         running.cancel("TIMEOUT");
+         return (cancelledStop());
       }
       try
       {
          AssistantTurn row = store.getTurn(userId, turn.getId());
          if(row != null && "CANCELLING".equals(row.getStatus()))
          {
-            running.cancel("USER");
+            running.cancel(row.getCancelReason() == null ? "USER" : row.getCancelReason());
             return (cancelledStop());
          }
       }
@@ -529,6 +674,18 @@ public class TurnEngine implements Runnable
     *******************************************************************************/
    private Stop cancelledStop()
    {
+      if("TIMEOUT".equals(running.getCancelReason()))
+      {
+         return (new Stop("ERROR", "TIMEOUT", TIMEOUT_TEXT, "timeout"));
+      }
+      if("SHUTDOWN".equals(running.getCancelReason()))
+      {
+         return (new Stop("ERROR", "INTERRUPTED", RESTART_TEXT, "interrupted"));
+      }
+      if("SUPERSEDED".equals(running.getCancelReason()))
+      {
+         return (new Stop("CANCELLED", null, SUPERSEDED_TEXT, ModelResponse.STOP_CANCELLED));
+      }
       if("KILL_SWITCH".equals(running.getCancelReason()))
       {
          String reason = StringUtils.hasContent(policy.killSwitchReason()) ? " " + policy.killSwitchReason() : "";
@@ -618,6 +775,7 @@ public class TurnEngine implements Runnable
          @Override
          public void onTextDelta(String text)
          {
+            phase("writing", null);
             bufferText(text);
          }
       };
@@ -639,6 +797,7 @@ public class TurnEngine implements Runnable
          TurnDataRouting.beforeModelCall(runtime, classification, turn, active);
          fallback.beforeCall(active);
          modelCallCount++;
+         phase("thinking", null);
          StreamHandle handle = provider.stream(request.copy(), fallback.listener(listener));
          running.setCurrentHandle(handle);
          try
@@ -701,12 +860,14 @@ public class TurnEngine implements Runnable
       for(int i = 0; i < toolUses.size(); i++)
       {
          ModelPart use = toolUses.get(i);
+         transcript.input(use.getInput());
          if(i >= allowed)
          {
             limitReached = true;
             continue;
          }
          toolCallCount++;
+         phase("tool", use.getToolName());
          emit(TurnEvent.TOOL_START, map("messageId", assistantMessage.getId(), "toolCallId", use.getToolUseId(), "name", use.getToolName(), "summary", ToolLabels.running(use.getToolName())));
 
          if(!allowedTools.contains(use.getToolName()))
@@ -720,8 +881,14 @@ public class TurnEngine implements Runnable
          AiToolCall   call  = new AiToolCall().withId(use.getToolUseId()).withName(use.getToolName()).withInput(use.getInput());
          futures.add(runtime.getExecutor().submit(() ->
          {
+            if(!transcript.retainWorker())
+            {
+               return;
+            }
+            transcript.attach();
             try
             {
+               QContext.init(runtime.getQInstance(), session);
                permits.acquire();
                try
                {
@@ -783,6 +950,12 @@ public class TurnEngine implements Runnable
                results[index] = new AiToolResult().withToolCallId(call.getId()).withToolName(call.getName()).withOutcome(AiToolOutcome.ERROR)
                   .withSummary("Failed").withModelContent("The tool failed unexpectedly.");
             }
+            finally
+            {
+               transcript.releaseWorker();
+               TranscriptRedactor.detach();
+               QContext.clear();
+            }
          }));
       }
       persistPendingEvents();
@@ -880,6 +1053,7 @@ public class TurnEngine implements Runnable
 
       try
       {
+         String inputJson = redactInput(PresentResultsTool.NAME.equals(use.getToolName()) ? (presentation ? result.getRender() : Map.of()) : use.getInput());
          store.insert(new AssistantToolCall()
             .withTurnId(turn.getId())
             .withConversationId(conversation.getId())
@@ -887,7 +1061,7 @@ public class TurnEngine implements Runnable
             .withToolCallId(use.getToolUseId())
             .withToolName(use.getToolName())
             .withClientChannel(scheduledTask == null ? "assistant/" + turn.getId() : "assistant/scheduled/" + scheduledTask.getId())
-            .withInputJson(truncate(redactInput(PresentResultsTool.NAME.equals(use.getToolName()) ? (presentation ? result.getRender() : Map.of()) : use.getInput()), MAX_INPUT_JSON_CHARS))
+            .withInputJson(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(inputJson) : inputJson)
             .withOutcome(outcome)
             .withPermissionCheck(result.getPermissionCheck())
             .withTableName(result.getTableName())
@@ -896,7 +1070,7 @@ public class TurnEngine implements Runnable
             .withRowCount(result.getRowCount())
             .withDurationMs(result.getDurationMs())
             .withResultSummary(summary)
-            .withResultJson(truncate(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(result.getModelContent()) : result.getModelContent(), MAX_RESULT_JSON_CHARS))
+            .withResultJson(BulkLoadTools.NAMES.contains(use.getToolName()) ? BulkLoadTools.forLog(result.getModelContent()) : result.getModelContent())
             .withRedactionsJson(result.getRedactions() == null || result.getRedactions().isEmpty() ? null : JsonUtils.toJson(result.getRedactions()))
             .withUiApiOperation(extras == null ? ("search_knowledge".equals(use.getToolName()) ? "GET /table/{tableName}/{primaryKey}" : null) : extras.getOperation())
             .withHttpStatus(extras == null ? null : extras.getHttpStatus())
@@ -996,93 +1170,103 @@ public class TurnEngine implements Runnable
    /*******************************************************************************
     ** Persist the message, turn, conversation and usage; emit the end events.
     *******************************************************************************/
-   private void finish(String status, String stopReason, String errorCode, String errorText)
+   private synchronized void finish(String status, String stopReason, String errorCode, String errorText)
    {
+      if(!running.tryFinalize())
+      {
+         return;
+      }
+      TranscriptRedactor previousTranscript = TranscriptRedactor.current();
       try
       {
-         persistClassification();
-         flushText(true);
-         Instant now = runtime.getClock().instant();
-
-         if(errorText != null)
-         {
-            parts.addNotice("error", errorText);
-         }
-
-         String model = this.model != null ? this.model : (policy == null ? null : policy.mainModel());
+         attachTranscript();
+         AssistantTurn persistedTurn = null;
          try
          {
-            store.updateFields(AssistantMessage.TABLE_NAME, new QRecord()
-               .withValue("id", assistantMessage.getId())
-               .withValue("status", status)
-               .withValue("model", model)
-               .withValue("partsJson", parts.toJson()));
-
-            AssistantConversation current = store.getConversation(userId, conversation.getId());
-            if(current != null)
-            {
-               store.updateFields(AssistantConversation.TABLE_NAME, new QRecord()
-                  .withValue("id", current.getId())
-                  .withValue("lastMessageAt", now)
-                  .withValue("unreadCount", (current.getUnreadCount() == null ? 0 : current.getUnreadCount()) + 1)
-                  .withValue("totalTokensIn", nz(current.getTotalTokensIn()) + nz(usage.getInputTokens()) + nz(usage.getCacheReadTokens()) + nz(usage.getCacheWriteTokens()))
-                  .withValue("totalTokensOut", nz(current.getTotalTokensOut()) + nz(usage.getOutputTokens()))
-                  .withValue("totalCostMicros", nz(current.getTotalCostMicros()) + costMicros));
-            }
-
-            store.addUsage(userId, LocalDate.now(runtime.getClock().withZone(ZoneOffset.UTC)), usage, costMicros, true);
+            persistedTurn = store.getTurn(userId, turn.getId());
          }
-         catch(QException e)
+         catch(Exception e)
          {
-            LOG.warn("Could not persist the end of an assistant turn", hipaaSafe(e), logPair("turnId", turn.getId()));
+            LOG.warn("Could not read final assistant status", hipaaSafe(e), logPair("turnId", turn.getId()));
+         }
+         boolean alreadyTerminal = persistedTurn != null && !("RUNNING".equals(persistedTurn.getStatus()) || "CANCELLING".equals(persistedTurn.getStatus()));
+         if(alreadyTerminal)
+         {
+            status = persistedTurn.getStatus();
+            stopReason = persistedTurn.getStopReason();
+            errorCode = persistedTurn.getErrorCode();
+         }
+         String finalStatus = status;
+         String finalStopReason = stopReason;
+         String finalErrorCode = errorCode;
+         String model = this.model != null ? this.model : (policy == null ? null : policy.mainModel());
+         Instant now = runtime.getClock().instant();
+         safely(() -> flushText(true));
+         if(errorText != null && !alreadyTerminal)
+         {
+            parts.addNotice("CANCELLED".equals(status) ? "info" : "error", errorText);
+         }
+         running.setPhase("done", null);
+         if(!alreadyTerminal)
+         {
+            safely(() -> persistClassification());
+            safely(() -> store.updateFields(AssistantMessage.TABLE_NAME, new QRecord()
+               .withValue("id", assistantMessage.getId()).withValue("status", finalStatus)
+               .withValue("model", model).withValue("partsJson", parts.toJson())));
          }
 
-         /////////////////////////////////////////////////////////////////////
-         // terminal events are persisted BEFORE the turn row turns final, //
-         // so a replay of a finished turn always ends with message.end    //
-         /////////////////////////////////////////////////////////////////////
          emit(TurnEvent.USAGE, map("turnId", turn.getId(), "inputTokens", usage.getInputTokens(), "outputTokens", usage.getOutputTokens(),
             "cacheReadTokens", usage.getCacheReadTokens(), "cacheWriteTokens", usage.getCacheWriteTokens(), "costMicros", costMicros));
          if(errorCode != null)
          {
             emit(TurnEvent.ERROR, map("code", errorCode, "message", errorText));
          }
-         Map<String, Object> end = map("messageId", assistantMessage.getId(), "status", status);
-         if(stopReason != null)
+         emit(TurnEvent.MESSAGE_END, map("messageId", assistantMessage.getId(), "status", status, "stopReason", stopReason));
+         if(!alreadyTerminal)
          {
-            end.put("stopReason", stopReason);
+            safely(this::persistPendingEvents);
+            safely(() -> transcript.flush(store));
+            // Accounting failures cannot skip the terminal row or local cleanup.
+            safely(() ->
+            {
+               store.updateRunningTurn(userId, new QRecord()
+                  .withValue("id", turn.getId())
+                  .withValue("status", finalStatus)
+                  .withValue("phase", "done")
+                  .withValue("stopReason", finalStopReason)
+                  .withValue("errorCode", finalErrorCode)
+                  .withValue("cancelReason", running.getCancelReason())
+                  .withValue("model", model)
+                  .withValue("providerId", active == null ? null : active.providerId())
+                  .withValue("routeDecision", turn.getRouteDecision())
+                  .withValue("routeSource", turn.getRouteSource())
+                  .withValue("fallbackCount", turn.getFallbackCount())
+                  .withValue("endedAt", now)
+                  .withValue("toolCallCount", toolCallCount)
+                  .withValue("modelCallCount", modelCallCount)
+                  .withValue("inputTokens", usage.getInputTokens())
+                  .withValue("outputTokens", usage.getOutputTokens())
+                  .withValue("cacheReadTokens", usage.getCacheReadTokens())
+                  .withValue("cacheWriteTokens", usage.getCacheWriteTokens())
+                  .withValue("costMicros", costMicros)
+                  .withValue("providerRequestIds", requestIds.isEmpty() ? null : String.join(",", requestIds)));
+            });
+            safely(() ->
+            {
+               AssistantConversation current = store.getConversation(userId, conversation.getId());
+               if(current != null)
+               {
+                  store.updateFields(AssistantConversation.TABLE_NAME, new QRecord()
+                     .withValue("id", current.getId())
+                     .withValue("lastMessageAt", now)
+                     .withValue("unreadCount", (current.getUnreadCount() == null ? 0 : current.getUnreadCount()) + 1)
+                     .withValue("totalTokensIn", nz(current.getTotalTokensIn()) + nz(usage.getInputTokens()) + nz(usage.getCacheReadTokens()) + nz(usage.getCacheWriteTokens()))
+                     .withValue("totalTokensOut", nz(current.getTotalTokensOut()) + nz(usage.getOutputTokens()))
+                     .withValue("totalCostMicros", nz(current.getTotalCostMicros()) + costMicros));
+               }
+            });
+            safely(() -> store.addUsage(userId, LocalDate.now(runtime.getClock().withZone(ZoneOffset.UTC)), usage, costMicros, true));
          }
-         emit(TurnEvent.MESSAGE_END, end);
-         persistPendingEvents();
-
-         try
-         {
-            store.updateFields(AssistantTurn.TABLE_NAME, new QRecord()
-               .withValue("id", turn.getId())
-               .withValue("status", status)
-               .withValue("stopReason", truncate(stopReason, 64))
-               .withValue("errorCode", errorCode)
-               .withValue("cancelReason", running.getCancelReason())
-               .withValue("model", model)
-               .withValue("providerId", active == null ? null : active.providerId())
-               .withValue("routeDecision", turn.getRouteDecision())
-               .withValue("routeSource", turn.getRouteSource())
-               .withValue("fallbackCount", turn.getFallbackCount())
-               .withValue("endedAt", now)
-               .withValue("toolCallCount", toolCallCount)
-               .withValue("modelCallCount", modelCallCount)
-               .withValue("inputTokens", usage.getInputTokens())
-               .withValue("outputTokens", usage.getOutputTokens())
-               .withValue("cacheReadTokens", usage.getCacheReadTokens())
-               .withValue("cacheWriteTokens", usage.getCacheWriteTokens())
-               .withValue("costMicros", costMicros)
-               .withValue("providerRequestIds", requestIds.isEmpty() ? null : String.join(",", requestIds)));
-         }
-         catch(QException e)
-         {
-            LOG.warn("Could not persist the end of an assistant turn", hipaaSafe(e), logPair("turnId", turn.getId()));
-         }
-
          LOG.info("Assistant turn finished",
             logPair("turnId", turn.getId()),
             logPair("conversationId", conversation.getId()),
@@ -1091,6 +1275,7 @@ public class TurnEngine implements Runnable
             logPair("status", status),
             logPair("stopReason", stopReason),
             logPair("errorCode", errorCode),
+            logPair("cancelReason", running.getCancelReason()),
             logPair("modelCalls", modelCallCount),
             logPair("toolCalls", toolCallCount),
             logPair("inputTokens", usage.getInputTokens()),
@@ -1100,29 +1285,104 @@ public class TurnEngine implements Runnable
             logPair("costMicros", costMicros),
             logPair("durationMs", Duration.between(startedAt, now).toMillis()));
       }
-      catch(QException | RuntimeException e)
-      {
-         LOG.warn("Assistant turn finish failed", hipaaSafe(e), logPair("turnId", turn.getId()));
-      }
       finally
       {
+         finished = true;
          running.getLog().close();
-         if(scheduledTask != null)
+         safely(() ->
          {
-            runtime.getScheduledTurn().detach(conversation.getId(), session.getUuid());
+            if(scheduledTask != null)
+            {
+               runtime.getScheduledTurn().detach(conversation.getId(), session.getUuid());
+            }
+         });
+         String finalStatus = status;
+         String finalText = errorText;
+         try
+         {
+            safely(() -> config.getRunTracker().finish(running.getTrackingId(), finalStatus, finalText));
+         }
+         finally
+         {
+            runtime.unregister(turn.getId());
+            runtime.getRateLimiter().release(userId);
+            if(transcript != null && !engineRunning)
+            {
+               transcript.close();
+            }
+            else
+            {
+               // A wedged provider can still log after forced finalization. The
+               // engine's finally releases its log protection when it returns.
+               TranscriptRedactor.detach();
+            }
+            if(previousTranscript != null && previousTranscript != transcript)
+            {
+               previousTranscript.attach();
+            }
          }
-         runtime.unregister(turn.getId());
-         runtime.getRateLimiter().release(userId);
       }
    }
 
 
 
+   /*******************************************************************************
+    ** One independently fallible finalization operation.
+    *******************************************************************************/
+   @FunctionalInterface
+   private interface FinishStep
+   {
+      /*******************************************************************************
+       ** Execute the step.
+       *******************************************************************************/
+      void run() throws Exception;
+   }
+
+
+
+   /*******************************************************************************
+    ** A failure in a collaborator must not strand the run.
+    *******************************************************************************/
+   private void safely(FinishStep step)
+   {
+      try
+      {
+         step.run();
+      }
+      catch(Exception e)
+      {
+         LOG.warn("Assistant lifecycle step failed", hipaaSafe(e), logPair("turnId", turn.getId()));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Emit and track only phase changes, including the current tool label.
+    *******************************************************************************/
+   private synchronized void phase(String phase, String toolName)
+   {
+      if(running.isFinalized() || (phase.equals(running.getPhase()) && Objects.equals(toolName, running.getToolName())))
+      {
+         return;
+      }
+      running.setPhase(phase, toolName);
+      String label = toolName == null ? phase : ToolLabels.running(toolName);
+      emit(TurnEvent.TURN_PHASE, map("phase", phase, "toolName", toolName, "label", label));
+      safely(() -> config.getRunTracker().progress(running.getTrackingId(), phase, label));
+   }
+
+
+
    /*******************************************************************************
     ** Buffer streamed text; emit it in small chunks (provider thread).
     *******************************************************************************/
-   private void bufferText(String text)
+   private synchronized void bufferText(String text)
    {
+      if(running.isFinalized())
+      {
+         return;
+      }
       synchronized(textBuffer)
       {
          textBuffer.append(text);
@@ -1135,7 +1395,7 @@ public class TurnEngine implements Runnable
    /*******************************************************************************
     ** Emit buffered text as one text.delta (when big or old enough, or forced).
     *******************************************************************************/
-   private void flushText(boolean force)
+   private synchronized void flushText(boolean force)
    {
       String chunk;
       synchronized(textBuffer)
@@ -1170,8 +1430,12 @@ public class TurnEngine implements Runnable
     ** Append to the live log; queue for persistence (done on the engine
     ** thread, which has the QContext).
     *******************************************************************************/
-   private void emit(String type, Map<String, Object> data)
+   private synchronized void emit(String type, Map<String, Object> data)
    {
+      if(finished)
+      {
+         return;
+      }
       toPersist.add(running.getLog().append(type, data));
    }
 
@@ -1180,18 +1444,44 @@ public class TurnEngine implements Runnable
    /*******************************************************************************
     ** Persist queued events.
     *******************************************************************************/
-   private void persistPendingEvents()
+   private synchronized void persistPendingEvents()
    {
+      if(!running.isFinalized() || finished)
+      {
+         return;
+      }
+      List<TurnEvent> events = new ArrayList<>();
       TurnEvent event;
       while((event = toPersist.poll()) != null)
+      {
+         events.add(event);
+      }
+      StringBuilder text = new StringBuilder();
+      for(TurnEvent queued : events)
+      {
+         if(TurnEvent.TEXT_DELTA.equals(queued.getType()))
+         {
+            text.append(queued.getData().get("text"));
+         }
+      }
+      Boolean firstText = true;
+      for(TurnEvent queued : events)
       {
          try
          {
-            store.insertEvent(turn.getId(), userId, event.getSeq(), event.getType(), event.toJson());
+            String json = queued.toJson();
+            if(TurnEvent.TEXT_DELTA.equals(queued.getType()))
+            {
+               Map<String, Object> data = new LinkedHashMap<>(queued.getData());
+               data.put("text", firstText ? text.toString() : "");
+               firstText = false;
+               json = JsonUtils.toJson(data);
+            }
+            store.insertEvent(turn.getId(), userId, queued.getSeq(), queued.getType(), json);
          }
-         catch(QException e)
+         catch(Exception e)
          {
-            LOG.warn("Could not persist assistant turn event", hipaaSafe(e), logPair("turnId", turn.getId()), logPair("seq", event.getSeq()));
+            LOG.warn("Could not persist assistant turn event", hipaaSafe(e), logPair("turnId", turn.getId()), logPair("seq", queued.getSeq()));
          }
       }
    }
@@ -1238,16 +1528,6 @@ public class TurnEngine implements Runnable
 
 
 
-   /*******************************************************************************
-    **
-    *******************************************************************************/
-   private static String truncate(String value, int max)
-   {
-      return (value == null || value.length() <= max ? value : value.substring(0, max));
-   }
-
-
-
    /*******************************************************************************
     **
     *******************************************************************************/

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/TurnEvent.java

@@ -29,6 +29,7 @@ import com.kingsrook.qqq.backend.core.utils.JsonUtils;
  *******************************************************************************/
 public class TurnEvent
 {
+   public static final String TURN_PHASE = "turn.phase";
    public static final String MESSAGE_START = "message.start";
    public static final String TEXT_DELTA    = "text.delta";
    public static final String TOOL_START    = "tool.start";

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/engine/UsageRecorder.java

@@ -17,6 +17,7 @@
 package com.kingsrook.qbits.assistant.engine;
 
 
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.model.AssistantUsage;
 import com.kingsrook.qbits.assistant.persistence.AssistantStore;
 import com.kingsrook.qbits.assistant.spi.ModelPricing;
@@ -145,6 +146,10 @@ public final class UsageRecorder
     *******************************************************************************/
    private static String truncate(String value, int max)
    {
+      if(value != null && value.length() > max && TranscriptRedactor.current() != null)
+      {
+         TranscriptRedactor.current().truncated(value, value.substring(0, max));
+      }
       return (value == null || value.length() <= max ? value : value.substring(0, max));
    }
 

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/jobs/AssistantRunTracker.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.jobs;
+
+
+/*******************************************************************************
+ ** STAND-IN(core=QRun-IO/qqq#1210, removeBy=RC.9): core job tracking
+ ** (jobs_job tracking rows) will implement this contract; hosts may wire their own.
+ *******************************************************************************/
+public interface AssistantRunTracker
+{
+   AssistantRunTracker NOOP = new AssistantRunTracker()
+   {
+      /*******************************************************************************
+       ** No tracking is configured.
+       *******************************************************************************/
+      @Override
+      public String start(String userId, Integer turnId, Integer conversationId, String label)
+      {
+         return (null);
+      }
+
+      /*******************************************************************************
+       ** No tracking is configured.
+       *******************************************************************************/
+      @Override
+      public void progress(String trackingId, String phase, String message)
+      {
+      }
+
+      /*******************************************************************************
+       ** No tracking is configured.
+       *******************************************************************************/
+      @Override
+      public void finish(String trackingId, String status, String message)
+      {
+      }
+   };
+
+   /*******************************************************************************
+    ** Publish a run and return its tracking id.
+    *******************************************************************************/
+   String start(String userId, Integer turnId, Integer conversationId, String label);
+
+   /*******************************************************************************
+    ** Publish a changed phase.
+    *******************************************************************************/
+   void progress(String trackingId, String phase, String message);
+
+   /*******************************************************************************
+    ** Publish the terminal outcome.
+    *******************************************************************************/
+   void finish(String trackingId, String status, String message);
+
+   /*******************************************************************************
+    ** Wire the host's Stop action to the owner's cancellation path.
+    *******************************************************************************/
+   default void onCancel(String trackingId, Runnable cancel)
+   {
+   }
+}

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantBudgetOverride.java

@@ -47,7 +47,7 @@ public class AssistantBudgetOverride extends QRecordEntity
    @QField()
    private Instant unlimitedUntil;
 
-   @QField(isRequired = true, maxLength = 1000)
+   @QField(isRequired = true)
    private String reason;
 
    @QField(isEditable = false, maxLength = 200)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantDataAudit.java

@@ -66,7 +66,7 @@ public class AssistantDataAudit extends QRecordEntity
    @QField(isEditable = false, maxLength = 16)
    private String decision;
 
-   @QField(isEditable = false, maxLength = 1000)
+   @QField(isEditable = false)
    private String reason;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantFeedback.java

@@ -22,7 +22,6 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.data.QField;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.data.QRecordEntity;
-import com.kingsrook.qqq.backend.core.model.metadata.fields.ValueTooLongBehavior;
 
 
 /*******************************************************************************
@@ -53,7 +52,7 @@ public class AssistantFeedback extends QRecordEntity
    @QField(isEditable = false)
    private Integer rating;
 
-   @QField(isEditable = false, maxLength = 64, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
+   @QField(isEditable = false)
    private String reason;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantPendingAction.java

@@ -22,7 +22,6 @@ import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.data.QField;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.data.QRecordEntity;
-import com.kingsrook.qqq.backend.core.model.metadata.fields.ValueTooLongBehavior;
 
 
 /*******************************************************************************
@@ -49,7 +48,7 @@ public class AssistantPendingAction extends QRecordEntity
    @QField(isEditable = false, maxLength = 255)
    private String userId;
 
-   @QField(isEditable = false, maxLength = 128)
+   @QField(isEditable = false)
    private String toolCallId;
 
    @QField(isEditable = false, maxLength = 64)
@@ -67,7 +66,7 @@ public class AssistantPendingAction extends QRecordEntity
    @QField(isEditable = false)
    private String paramsJson;
 
-   @QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
+   @QField(isEditable = false)
    private String summary;
 
    @QField(isEditable = false, maxLength = 16)
@@ -76,7 +75,7 @@ public class AssistantPendingAction extends QRecordEntity
    @QField(isEditable = false)
    private Integer httpStatus;
 
-   @QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
+   @QField(isEditable = false)
    private String resultSummary;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantScheduledRun.java

@@ -62,13 +62,13 @@ public class AssistantScheduledRun extends QRecordEntity
    @QField()
    private Integer turnId;
 
-   @QField(maxLength = 2000)
+   @QField()
    private String summary;
 
    @QField()
    private Long costMicros;
 
-   @QField(maxLength = 500)
+   @QField()
    private String error;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantToolCall.java

@@ -55,7 +55,7 @@ public class AssistantToolCall extends QRecordEntity
    @QField(isEditable = false, maxLength = 255)
    private String userId;
 
-   @QField(isEditable = false, maxLength = 128)
+   @QField(isEditable = false)
    private String toolCallId;
 
    @QField(isEditable = false, maxLength = 64)
@@ -85,7 +85,7 @@ public class AssistantToolCall extends QRecordEntity
    @QField(isEditable = false)
    private Long durationMs;
 
-   @QField(isEditable = false, maxLength = 500, valueTooLongBehavior = ValueTooLongBehavior.TRUNCATE_ELLIPSIS)
+   @QField(isEditable = false)
    private String resultSummary;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantTurn.java

@@ -74,7 +74,7 @@ public class AssistantTurn extends QRecordEntity
    @QField(isEditable = false, maxLength = 16)
    private String status;
 
-   @QField(isEditable = false, maxLength = 64)
+   @QField(isEditable = false)
    private String stopReason;
 
    @QField(isEditable = false, maxLength = 32)
@@ -92,6 +92,21 @@ public class AssistantTurn extends QRecordEntity
    @QField(isEditable = false)
    private Instant endedAt;
 
+   @QField(isEditable = false)
+   private Instant heartbeatAt;
+
+   @QField(isEditable = false, maxLength = 100)
+   private String leaseOwner;
+
+   @QField(isEditable = false)
+   private Instant cancelRequestedAt;
+
+   @QField(isEditable = false, maxLength = 64)
+   private String phase;
+
+   @QField(isEditable = false, maxLength = 16)
+   private String concurrencyMode;
+
    @QField(isEditable = false)
    private Integer toolCallCount;
 
@@ -1011,4 +1026,160 @@ public class AssistantTurn extends QRecordEntity
       this.providerId = providerId;
       return (this);
    }
+
+
+
+   /*******************************************************************************
+    ** Getter for heartbeatAt
+    *******************************************************************************/
+   public Instant getHeartbeatAt()
+   {
+      return (this.heartbeatAt);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for heartbeatAt
+    *******************************************************************************/
+   public void setHeartbeatAt(Instant heartbeatAt)
+   {
+      this.heartbeatAt = heartbeatAt;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for heartbeatAt
+    *******************************************************************************/
+   public AssistantTurn withHeartbeatAt(Instant heartbeatAt)
+   {
+      this.heartbeatAt = heartbeatAt;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for leaseOwner
+    *******************************************************************************/
+   public String getLeaseOwner()
+   {
+      return (this.leaseOwner);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for leaseOwner
+    *******************************************************************************/
+   public void setLeaseOwner(String leaseOwner)
+   {
+      this.leaseOwner = leaseOwner;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for leaseOwner
+    *******************************************************************************/
+   public AssistantTurn withLeaseOwner(String leaseOwner)
+   {
+      this.leaseOwner = leaseOwner;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for cancelRequestedAt
+    *******************************************************************************/
+   public Instant getCancelRequestedAt()
+   {
+      return (this.cancelRequestedAt);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for cancelRequestedAt
+    *******************************************************************************/
+   public void setCancelRequestedAt(Instant cancelRequestedAt)
+   {
+      this.cancelRequestedAt = cancelRequestedAt;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for cancelRequestedAt
+    *******************************************************************************/
+   public AssistantTurn withCancelRequestedAt(Instant cancelRequestedAt)
+   {
+      this.cancelRequestedAt = cancelRequestedAt;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for phase
+    *******************************************************************************/
+   public String getPhase()
+   {
+      return (this.phase);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for phase
+    *******************************************************************************/
+   public void setPhase(String phase)
+   {
+      this.phase = phase;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for phase
+    *******************************************************************************/
+   public AssistantTurn withPhase(String phase)
+   {
+      this.phase = phase;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for concurrencyMode
+    *******************************************************************************/
+   public String getConcurrencyMode()
+   {
+      return (this.concurrencyMode);
+   }
+
+
+
+   /*******************************************************************************
+    ** Setter for concurrencyMode
+    *******************************************************************************/
+   public void setConcurrencyMode(String concurrencyMode)
+   {
+      this.concurrencyMode = concurrencyMode;
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for concurrencyMode
+    *******************************************************************************/
+   public AssistantTurn withConcurrencyMode(String concurrencyMode)
+   {
+      this.concurrencyMode = concurrencyMode;
+      return (this);
+   }
+
 }

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/model/AssistantUsage.java

@@ -84,7 +84,7 @@ public class AssistantUsage extends QRecordEntity
    @QField(isEditable = false)
    private Long costMicros;
 
-   @QField(isEditable = false, maxLength = 100)
+   @QField(isEditable = false)
    private String requestId;
 
    @QField(isEditable = false)

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/persistence/AssistantStore.java

@@ -31,6 +31,7 @@ import com.kingsrook.qbits.assistant.AssistantQBitConfig;
 import com.kingsrook.qbits.assistant.AssistantRuntime;
 import com.kingsrook.qbits.assistant.classification.Classifications;
 import com.kingsrook.qbits.assistant.engine.TurnClassification;
+import com.kingsrook.qbits.assistant.engine.TurnEvent;
 import com.kingsrook.qbits.assistant.model.AssistantBudget;
 import com.kingsrook.qbits.assistant.model.AssistantBudgetEvent;
 import com.kingsrook.qbits.assistant.model.AssistantBudgetOverride;
@@ -161,7 +162,13 @@ public class AssistantStore
       {
          projectBudgetUsage(record);
       }
-      return ((T) QRecordEntity.fromQRecord(entity.getClass(), cryptRecord(entity.tableName(), record, false)));
+      rememberTranscript(entity.tableName(), entity.toQRecord(), record.getValue("id"));
+      QRecord returned = cryptRecord(entity.tableName(), record, false);
+      if(TranscriptRedactor.current() != null)
+      {
+         TranscriptRedactor.current().restore(returned, entity.toQRecord());
+      }
+      return ((T) QRecordEntity.fromQRecord(entity.getClass(), returned));
    }
 
 
@@ -172,6 +179,7 @@ public class AssistantStore
    public void update(QRecordEntity entity) throws QException
    {
       requireOwnTable(entity.tableName());
+      rememberTranscript(entity.tableName(), entity.toQRecord(), entity.toQRecord().getValue("id"));
       UpdateOutput output = new UpdateAction().execute(new UpdateInput(entity.tableName()).withRecord(cryptRecord(entity.tableName(), entity.toQRecord(), true)));
       QRecord      record = output.getRecords().get(0);
       if(CollectionUtils.nullSafeHasContents(record.getErrors()))
@@ -188,6 +196,7 @@ public class AssistantStore
    public void updateFields(String tableName, QRecord record) throws QException
    {
       requireOwnTable(tableName);
+      rememberTranscript(tableName, record, record.getValue("id"));
       UpdateOutput output = new UpdateAction().execute(new UpdateInput(tableName).withRecord(cryptRecord(tableName, record, true)));
       QRecord      result = output.getRecords().get(0);
       if(CollectionUtils.nullSafeHasContents(result.getErrors()))
@@ -384,6 +393,27 @@ public class AssistantStore
 
 
 
+   /*******************************************************************************
+    ** Re-read before a lifecycle update. A completed row is never reopened by
+    ** a late local Stop or overwritten by a second local finalizer. Every read
+    ** and update remains under the caller's own session and owner filter.
+    *******************************************************************************/
+   public boolean updateRunningTurn(String userId, QRecord changes) throws QException
+   {
+      synchronized(QContext.getQInstance())
+      {
+         AssistantTurn current = getTurn(userId, changes.getValueInteger("id"));
+         if(current == null || !("RUNNING".equals(current.getStatus()) || "CANCELLING".equals(current.getStatus())))
+         {
+            return (false);
+         }
+         updateFields(AssistantTurn.TABLE_NAME, changes);
+         return (true);
+      }
+   }
+
+
+
    /*******************************************************************************
     ** One of the user's own pending actions, or null.
     *******************************************************************************/
@@ -411,6 +441,51 @@ public class AssistantStore
     ** Persist one streamed event.
     *******************************************************************************/
    public void insertEvent(Integer turnId, String userId, int seq, String eventType, String dataJson) throws QException
+   {
+      if(TurnEvent.MESSAGE_END.equals(eventType) || TurnEvent.ERROR.equals(eventType))
+      {
+         // Serialize memory-backend callers. The Postgres partial unique index
+         // arbitrates this same race between independent replicas.
+         synchronized(QContext.getQInstance())
+         {
+            if(hasTerminalEvent(userId, turnId))
+            {
+               return;
+            }
+            try
+            {
+               insertStreamEvent(turnId, userId, seq, eventType, dataJson);
+            }
+            catch(QException e)
+            {
+               if(!hasTerminalEvent(userId, turnId))
+               {
+                  throw (e);
+               }
+            }
+         }
+         return;
+      }
+      insertStreamEvent(turnId, userId, seq, eventType, dataJson);
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether a terminal replay frame has already been persisted by any owner.
+    *******************************************************************************/
+   private boolean hasTerminalEvent(String userId, Integer turnId) throws QException
+   {
+      return (!query(AssistantTurnEvent.TABLE_NAME, owned(userId).withCriteria("turnId", QCriteriaOperator.EQUALS, turnId)
+         .withCriteria("eventType", QCriteriaOperator.EQUALS, TurnEvent.MESSAGE_END).withLimit(1), AssistantTurnEvent.class).isEmpty());
+   }
+
+
+
+   /*******************************************************************************
+    ** Persist one frame through the normal encrypted, user-owned store path.
+    *******************************************************************************/
+   private void insertStreamEvent(Integer turnId, String userId, int seq, String eventType, String dataJson) throws QException
    {
       insert(new AssistantTurnEvent()
          .withTurnId(turnId)
@@ -832,6 +907,10 @@ public class AssistantStore
       }
       QRecord copy = new QRecord();
       copy.setValues(new LinkedHashMap<>(record.getValues()));
+      if(encrypt && TranscriptRedactor.current() != null)
+      {
+         copy = TranscriptRedactor.current().prepare(copy);
+      }
       for(String field : ENCRYPTED_FIELDS.getOrDefault(tableName, List.of()))
       {
          if(copy.getValue(field) instanceof String text)
@@ -844,6 +923,19 @@ public class AssistantStore
 
 
 
+   /*******************************************************************************
+    ** Stage transcript payloads until all classified values of the turn are known.
+    *******************************************************************************/
+   private void rememberTranscript(String table, QRecord record, Serializable id)
+   {
+      if(TranscriptRedactor.current() != null)
+      {
+         TranscriptRedactor.current().saved(table, record, id);
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Retention candidates are paged independently of deletion, so retained PHI cannot starve other rows.
     *******************************************************************************/

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/persistence/TranscriptRedactor.java

@@ -0,0 +1,495 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.persistence;
+
+
+import java.io.Serializable;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import com.kingsrook.qbits.assistant.classification.ClassificationSource;
+import com.kingsrook.qbits.assistant.model.AssistantPendingAction;
+import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
+import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueTokens;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.LogRedactorInterface;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
+
+
+/*******************************************************************************
+ ** Turn-local storage egress protection. Values and provisional payloads live only
+ ** in memory; QQQ actions receive placeholders until the complete turn is known.
+ ** The log hook is global while active, covering provider and tool worker threads.
+ *******************************************************************************/
+public class TranscriptRedactor implements LogRedactorInterface, AutoCloseable
+{
+   private static final ThreadLocal<TranscriptRedactor> CURRENT = new ThreadLocal<>();
+   private static final Set<String> PAYLOAD_FIELDS = Set.of("partsJson", "inputJson", "resultJson", "redactionsJson", "dataJson", "paramsJson",
+      "summary", "resultSummary", "reason", "title", "contextJson", "errorMessage", "errorText", "error", "requestId", "comment", "providerRequestId", "providerRequestIds", "auditHref", "toolCallId", "stopReason");
+   private final ClassificationSource classifications;
+   private final QInstance instance;
+   private final Map<String, String> replacements = new LinkedHashMap<>();
+   private final Map<List<String>, String> recordTokens = new LinkedHashMap<>();
+   private final Map<List<String>, QRecord> pending = new LinkedHashMap<>();
+   private Boolean flushing = false;
+   private volatile Boolean closed = false;
+   private volatile Boolean released = false;
+   private Integer workers = 0;
+
+
+
+   /*******************************************************************************
+    ** Register only for this turn; close releases both values and the log hook.
+    *******************************************************************************/
+   public TranscriptRedactor(QInstance instance, ClassificationSource classifications)
+   {
+      this.instance = instance;
+      this.classifications = classifications;
+      QLogger.registerRedactor(this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Explicit propagation works for pooled tool executors, without inheritable state.
+    *******************************************************************************/
+   public void attach()
+   {
+      CURRENT.set(this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Remove only the thread binding, retaining the turn's in-memory state.
+    *******************************************************************************/
+   public static void detach()
+   {
+      CURRENT.remove();
+   }
+
+
+
+   /*******************************************************************************
+    ** Storage and tools share the currently bound turn, if any.
+    *******************************************************************************/
+   public static TranscriptRedactor current()
+   {
+      return (CURRENT.get());
+   }
+
+
+
+   /*******************************************************************************
+    ** Capture the raw canonical value and the actual displayed/truncated variant.
+    ** Tokens remain bound to the original record field, including equal values.
+    *******************************************************************************/
+   public void observe(String tableName, String fieldName, Serializable primaryKey, Object value, Object displayed)
+   {
+      QTableMetaData table = instance.getTable(tableName);
+      if(released || value == null || table == null || table.getField(fieldName) == null
+         || !classifications.ofField(tableName, fieldName).isAtLeast(DataClassification.PII))
+      {
+         return;
+      }
+      Serializable original = value instanceof Serializable serializable ? serializable : String.valueOf(value);
+      String token = primaryKey == null
+         ? ClassifiedValueTokens.forContext(List.of("assistant", tableName, fieldName), original)
+         : ClassifiedValueTokens.forRecordField(table, table.getField(fieldName), primaryKey, original);
+      synchronized(this)
+      {
+         if(released)
+         {
+            return;
+         }
+         recordTokens.put(List.of(tableName, fieldName, String.valueOf(primaryKey), String.valueOf(displayed)), token);
+         remember(ClassifiedValueTokens.canonical(original), token);
+         remember(String.valueOf(value), token);
+         if(displayed != null)
+         {
+            remember(String.valueOf(displayed), token);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A snippet is another spelling of an already observed field, not a new value.
+    *******************************************************************************/
+   public synchronized void variant(String table, String field, String primaryKey, String text)
+   {
+      if(!closed && text != null && table != null && field != null && primaryKey != null)
+      {
+         recordTokens.entrySet().stream().filter(entry -> entry.getKey().subList(0, 3).equals(List.of(table, field, primaryKey)))
+            .findFirst().ifPresent(entry -> remember(text, entry.getValue()));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Also protect JSON-escaped and URL-encoded occurrences in envelopes and links.
+    *******************************************************************************/
+   private void remember(String value, String token)
+   {
+      if(value == null || value.isEmpty())
+      {
+         return;
+      }
+      replacements.putIfAbsent(value, token);
+      String json = JsonUtils.toJson(value);
+      replacements.putIfAbsent(json.substring(1, json.length() - 1), token);
+      replacements.putIfAbsent(URLEncoder.encode(value, StandardCharsets.UTF_8), token);
+   }
+
+
+
+   /*******************************************************************************
+    ** Existing tool summaries may clip a value before storage. Remember that exact
+    ** prefix as another variant while keeping the live/model summary unchanged.
+    *******************************************************************************/
+   public synchronized void truncated(String original, String prefix)
+   {
+      for(var entry : new ArrayList<>(replacements.entrySet()))
+      {
+         int start = original.indexOf(entry.getKey());
+         while(start >= 0)
+         {
+            if(start < prefix.length() && start + entry.getKey().length() > prefix.length())
+            {
+               remember(prefix.substring(start), entry.getValue());
+            }
+            start = original.indexOf(entry.getKey(), start + 1);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Criteria and proposed values are classified even when a query returns no rows.
+    *******************************************************************************/
+   public void input(Map<String, Object> input)
+   {
+      if(input != null && input.get("table") instanceof String table)
+      {
+         criteria(table, input.get("filter"));
+         if(input.get("values") instanceof Map<?, ?> values)
+         {
+            values.forEach((field, value) -> observe(table, String.valueOf(field), null, value, value));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Traverse both model and QQQ filter spellings, including nested boolean groups.
+    *******************************************************************************/
+   private void criteria(String table, Object node)
+   {
+      if(node instanceof Map<?, ?> map)
+      {
+         Object field = map.containsKey("field") ? map.get("field") : map.get("fieldName");
+         if(field instanceof String name && map.get("values") instanceof List<?> values)
+         {
+            values.forEach(value -> observe(table, name, null, value, value));
+         }
+         map.values().forEach(value -> criteria(table, value));
+      }
+      else if(node instanceof List<?> list)
+      {
+         list.forEach(value -> criteria(table, value));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Longest exact match wins; a single pass never redacts a generated token again.
+    *******************************************************************************/
+   @Override
+   public synchronized String redact(String text)
+   {
+      if(text == null || replacements.isEmpty())
+      {
+         return (text);
+      }
+      List<String> values = new ArrayList<>(replacements.keySet());
+      values.sort(Comparator.comparingInt(String::length).reversed());
+      Pattern pattern = Pattern.compile("(?:hmac|shred):v1:[A-Za-z0-9._-]+:[A-Za-z0-9_-]+|changed:v1|" + String.join("|", values.stream().map(Pattern::quote).toList()));
+      Matcher matcher = pattern.matcher(text);
+      return (matcher.replaceAll(match -> Matcher.quoteReplacement(replacements.getOrDefault(match.group(), match.group()))));
+   }
+
+
+
+   /*******************************************************************************
+    ** Redact JSON leaves without damaging property names or numeric syntax. Enveloped
+    ** tool data retains its framing. Invalid/opaque content is still exact-redacted.
+    *******************************************************************************/
+   public String payload(String text)
+   {
+      if(text == null)
+      {
+         return (null);
+      }
+      int start = text.indexOf('{');
+      int end = text.lastIndexOf('}');
+      try
+      {
+         if(text.stripLeading().startsWith("["))
+         {
+            return (JsonUtils.toJson(tree(JsonUtils.toObject(text, Object.class), null, null)));
+         }
+         if(start >= 0 && end >= start)
+         {
+            Object json = JsonUtils.toObject(text.substring(start, end + 1), Object.class);
+            return (redact(text.substring(0, start)) + JsonUtils.toJson(tree(json, null, null)) + redact(text.substring(end + 1)));
+         }
+      }
+      catch(Exception e)
+      {
+         return (redact(text));
+      }
+      return (redact(text));
+   }
+
+
+
+   /*******************************************************************************
+    ** Preserve record bindings in structured query/get results and render copies.
+    *******************************************************************************/
+   private synchronized Object tree(Object node, String table, String primaryKey)
+   {
+      if(node instanceof Map<?, ?> map)
+      {
+         Object namedTable = map.containsKey("table") ? map.get("table") : map.get("tableName");
+         String resolvedTable = namedTable instanceof String name ? name : table;
+         Object id = map.containsKey("id") ? map.get("id") : map.get("recordId");
+         String resolvedKey = id == null ? primaryKey : String.valueOf(id);
+         Map<String, Object> copy = new LinkedHashMap<>();
+         map.forEach((key, value) ->
+         {
+            String token = recordTokens.get(List.of(String.valueOf(resolvedTable), String.valueOf(key), String.valueOf(resolvedKey), String.valueOf(value)));
+            copy.put(redact(String.valueOf(key)), token == null ? tree(value, resolvedTable, resolvedKey) : token);
+         });
+         return (copy);
+      }
+      if(node instanceof List<?> list)
+      {
+         return (list.stream().map(value -> tree(value, table, primaryKey)).toList());
+      }
+      if(node instanceof String text)
+      {
+         return (redact(text));
+      }
+      if(node instanceof Number && replacements.containsKey(String.valueOf(node)))
+      {
+         return (replacements.get(String.valueOf(node)));
+      }
+      return (node);
+   }
+
+
+
+   /*******************************************************************************
+    ** Never put provisional content in a backend, its audits, or its SQL logs.
+    *******************************************************************************/
+   public synchronized QRecord prepare(QRecord original)
+   {
+      QRecord copy = new QRecord();
+      copy.setValues(new LinkedHashMap<>(original.getValues()));
+      for(String field : PAYLOAD_FIELDS)
+      {
+         if(copy.getValue(field) instanceof String text)
+         {
+            copy.setValue(field, flushing && !closed ? stored(field, text) : (field.endsWith("Json") ? (field.equals("partsJson") ? "[]" : "{}") : ClassifiedValueTokens.changed()));
+         }
+      }
+      if(!flushing && copy.getValues().containsKey("paramsJson") && !AssistantPendingAction.STATUS_REJECTED.equals(copy.getValueString("status")))
+      {
+         copy.setValue("status", "PREPARING");
+      }
+      return (copy);
+   }
+
+
+
+   /*******************************************************************************
+    ** Truncate only after redaction, so a storage bound never cuts a raw value.
+    *******************************************************************************/
+   private String stored(String field, String text)
+   {
+      String safe = payload(text);
+      Integer limit = switch(field)
+      {
+         case "inputJson" -> 4000;
+         case "resultJson" -> 64000;
+         case "title" -> 60;
+         default -> null;
+      };
+      if("title".equals(field))
+      {
+         safe = safe.strip().split("\\R", 2)[0].replaceAll("\\s+", " ").strip();
+      }
+      return (limit != null && safe.length() > limit ? safe.substring(0, limit - 1) + "…" : safe);
+   }
+
+
+
+   /*******************************************************************************
+    ** Return live content to the caller without losing backend-generated defaults.
+    *******************************************************************************/
+   public void restore(QRecord returned, QRecord original)
+   {
+      for(String field : PAYLOAD_FIELDS)
+      {
+         if(original.getValues().containsKey(field))
+         {
+            returned.setValue(field, original.getValue(field));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Remember only content columns, merging later writes without stale bookkeeping.
+    *******************************************************************************/
+   public synchronized void saved(String table, QRecord original, Serializable id)
+   {
+      if(flushing || closed || PAYLOAD_FIELDS.stream().noneMatch(original.getValues()::containsKey))
+      {
+         return;
+      }
+      QRecord content = pending.computeIfAbsent(List.of(table, String.valueOf(id)), key -> new QRecord().withValue("id", id));
+      for(String field : PAYLOAD_FIELDS)
+      {
+         if(original.getValues().containsKey(field))
+         {
+            content.setValue(field, original.getValue(field));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Flush only redacted copies. A crash/failure leaves value-free placeholders.
+    ** Redacted proposals cannot later submit token strings to an application field.
+    *******************************************************************************/
+   public void flush(AssistantStore store) throws QException
+   {
+      Map<List<String>, QRecord> writes;
+      Boolean complete;
+      synchronized(this)
+      {
+         complete = workers == 0;
+         flushing = complete;
+         writes = new LinkedHashMap<>(pending);
+         pending.clear();
+      }
+      for(var entry : writes.entrySet())
+      {
+         QRecord record = entry.getValue();
+         String table = entry.getKey().getFirst();
+         if(AssistantPendingAction.TABLE_NAME.equals(table) && record.getValueString("paramsJson") != null)
+         {
+            Boolean changed = !complete || !record.getValueString("paramsJson").equals(payload(record.getValueString("paramsJson")));
+            record.setValue("status", changed ? AssistantPendingAction.STATUS_REJECTED : AssistantPendingAction.STATUS_PENDING);
+            if(changed)
+            {
+               record.setValue("resultSummary", "Classified values were removed. Make this change in the application.");
+            }
+         }
+         store.updateFields(table, record);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A timed-out tool can finish after the engine; retain its log protection.
+    *******************************************************************************/
+   public synchronized Boolean retainWorker()
+   {
+      if(closed)
+      {
+         return (false);
+      }
+      workers++;
+      return (true);
+   }
+
+
+
+   /*******************************************************************************
+    ** Release a completed worker.
+    *******************************************************************************/
+   public synchronized void releaseWorker()
+   {
+      workers--;
+      releaseIfDone();
+   }
+
+
+
+   /*******************************************************************************
+    ** Clear classified values only after the engine and all tool workers finish.
+    *******************************************************************************/
+   private void releaseIfDone()
+   {
+      if(closed && workers == 0)
+      {
+         released = true;
+         QLogger.unregisterRedactor(this);
+         replacements.clear();
+         recordTokens.clear();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Always release raw data and unregister, including failed/cancelled turns.
+    *******************************************************************************/
+   @Override
+   public synchronized void close()
+   {
+      closed = true;
+      pending.clear();
+      releaseIfDone();
+      detach();
+   }
+}

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/KnowledgeTool.java

@@ -25,6 +25,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import com.kingsrook.qbits.assistant.AssistantRuntime;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.engine.PolicyService;
 import com.kingsrook.qbits.assistant.engine.TurnEngine;
 import com.kingsrook.qbits.assistant.model.AssistantRagSource;
@@ -213,6 +214,11 @@ public class KnowledgeTool
             title = title.substring(0, Math.min(300, title.length()));
             String snippet = chunks.get(hit.chunkIndex());
             snippet = snippet.substring(0, Math.min(1500, snippet.length()));
+            if(TranscriptRedactor.current() != null)
+            {
+               TranscriptRedactor.current().variant(hit.table(), hit.field(), hit.pk(), snippet);
+               TranscriptRedactor.current().variant(hit.table(), source.getTitleFieldName(), hit.pk(), title);
+            }
             records.add(Map.of("table", hit.table(), "recordId", hit.pk(), "title", title, "snippet", snippet, "score", hit.score(), "href", href));
             result.getRecordRefs().add(new AiRecordRef().withTableName(hit.table()).withId(hit.pk()).withLabel(title).withHref(href));
             result.getRedactions().addAll(read.getResult().getRedactions());

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/UiApiToolbox.java

@@ -27,6 +27,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.access.DataAccessMode;
 import com.kingsrook.qbits.assistant.access.DataAccessService;
 import com.kingsrook.qbits.assistant.engine.PolicyService;
@@ -1445,7 +1446,12 @@ public class UiApiToolbox
          {
             text = AiValueMasker.mask(text, field.getType());
          }
-         rs.put(name, sanitize(text, context, redactions));
+         String visible = sanitize(text, context, redactions);
+         if(TranscriptRedactor.current() != null)
+         {
+            TranscriptRedactor.current().observe(table.getName(), name, (Serializable) raw.get(table.getPrimaryKeyField()), entry.getValue(), visible);
+         }
+         rs.put(name, visible);
       }
       return (rs);
    }
@@ -1534,6 +1540,10 @@ public class UiApiToolbox
          return ("");
       }
       String line = text.replaceAll("\\s+", " ").strip();
+      if(line.length() > 300 && TranscriptRedactor.current() != null)
+      {
+         TranscriptRedactor.current().truncated(line, line.substring(0, 299));
+      }
       return (line.length() <= 300 ? line : line.substring(0, 299) + "…");
    }
 

qbit-assistant/src/main/java/com/kingsrook/qbits/assistant/tools/bulkload/BulkLoadTools.java

@@ -23,6 +23,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
 import com.kingsrook.qbits.assistant.access.DataAccessService;
 import com.kingsrook.qbits.assistant.engine.PolicyService;
 import com.kingsrook.qbits.assistant.spi.ModelToolDefinition;
@@ -823,6 +824,10 @@ public class BulkLoadTools
       {
          return ("");
       }
+      if(text.length() > 300 && TranscriptRedactor.current() != null)
+      {
+         TranscriptRedactor.current().truncated(text, text.substring(0, 297));
+      }
       return (text.length() > 300 ? text.substring(0, 297) + "..." : text);
    }
 

qbit-assistant/src/main/java/com/kingsrook/qqq/middleware/javalin/specs/invoke/HeaderedUiApiInvoker.java

@@ -83,7 +83,7 @@ public class HeaderedUiApiInvoker extends UiApiInvoker
       {
          if(!RESERVED_HEADERS.contains(name.toLowerCase(Locale.ROOT)))
          {
-            builder.header(name, value);
+            builder.setHeader(name, value);
          }
       });
       return (builder.build());

qbit-assistant/src/main/resources/openapi/qqq-v1/assistant.yaml

@@ -53,6 +53,43 @@ components:
           nullable: true
           type: "string"
       type: "object"
+    AssistantActiveTurnsResponseV1:
+      description: "The caller's active runs, newest first; stale leases are reaped."
+      properties:
+        turns:
+          items:
+            description: "An active assistant run."
+            properties:
+              turnId:
+                type: "integer"
+              conversationId:
+                type: "integer"
+              assistantMessageId:
+                type: "integer"
+              userMessageId:
+                type: "integer"
+              status:
+                description: "RUNNING or CANCELLING."
+                type: "string"
+              phase:
+                type: "string"
+              startedAt:
+                description: "ISO-8601."
+                type: "string"
+              heartbeatAt:
+                description: "ISO-8601."
+                type: "string"
+              elapsedMs:
+                type: "integer"
+              live:
+                description: "Running on this replica."
+                type: "boolean"
+              concurrencyMode:
+                description: "SUPERSEDE or CONCURRENT."
+                type: "string"
+            type: "object"
+          type: "array"
+      type: "object"
     AssistantConfigResponseV1:
       description: "Whether the assistant is available to this user, and how."
       properties:
@@ -187,8 +224,8 @@ components:
       type: "object"
     AssistantTurnEventV1:
       description: "One server-sent event of a turn (event: <type>, id: <seq>, data:\
-        \ this object). Types: message.start, text.delta, tool.start, tool.result,\
-        \ ui.navigate, notice, usage, message.end, error, ping."
+        \ this object). Types: message.start, turn.phase, text.delta, tool.start,\
+        \ tool.result, ui.navigate, notice, usage, message.end, error, ping."
       properties:
         seq:
           description: "Sequence number; resume with ?after=<seq>."
@@ -198,7 +235,9 @@ components:
           type: "string"
         data:
           additionalProperties: true
-          description: "The event's fields, by type."
+          description: "The event's fields, by type. message.start includes startedAt\
+            \ (ISO-8601). turn.phase includes phase (thinking, tool, writing), optional\
+            \ toolName and label."
           type: "object"
       type: "object"
   securitySchemes:
@@ -212,7 +251,7 @@ info:
   description: "QQQ Assistant: conversations, streamed turns, feedback, pending actions\
     \ and data access explanations"
   title: "QQQ Middleware API extension: assistant"
-  version: "1.1.0"
+  version: "1.2.0"
 openapi: "3.0.3"
 paths:
   /qqq/v1/assistant/config:
@@ -605,6 +644,10 @@ paths:
                   additionalProperties: true
                   description: "Where the user is in the app."
                   type: "object"
+                concurrency:
+                  description: "supersede or concurrent (case-insensitive); defaults\
+                    \ to the host configuration."
+                  type: "string"
               type: "object"
         required: true
       responses:
@@ -754,6 +797,12 @@ paths:
               schema:
                 $ref: "#/components/schemas/AssistantErrorV1"
           description: "Disabled, not configured, or the kill switch is on"
+        200:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantStatusResponseV1"
+          description: "Successful Response"
         409:
           content:
             application/json:
@@ -765,7 +814,7 @@ paths:
             application/json:
               schema:
                 $ref: "#/components/schemas/AssistantStatusResponseV1"
-          description: "Successful Response"
+          description: "Legacy replica: cancellation accepted"
         429:
           content:
             application/json:
@@ -776,6 +825,56 @@ paths:
       - sessionUuidCookie:
         - "N/A"
       summary: "Stop a running turn"
+  /qqq/v1/assistant/turns/active:
+    get:
+      description: "Find the user's active assistant runs"
+      responses:
+        400:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "Invalid input"
+        403:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "No access to the assistant"
+        404:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "Not found (or not the user's own)"
+        503:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "Disabled, not configured, or the kill switch is on"
+        200:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantActiveTurnsResponseV1"
+          description: "Successful Response"
+        409:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "Already finished or decided"
+        429:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/AssistantErrorV1"
+          description: "Budget or rate limit"
+      security:
+      - sessionUuidCookie:
+        - "N/A"
+      summary: "Find the user's active assistant runs"
   /qqq/v1/assistant/messages/{id}/feedback:
     post:
       description: "Thumbs up or down on an assistant message"

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/AssistantDefaultsChangelogTest.java

@@ -345,6 +345,11 @@ class AssistantDefaultsChangelogTest extends BaseAssistantTest
       {
          for(String sql : statements(ddl))
          {
+            if(sql.startsWith("DO $$"))
+            {
+               // the foreign-key block is PL/pgSQL (Postgres only); SchemaForeignKeyPostgresTest runs it on real Postgres
+               continue;
+            }
             Matcher m = Pattern.compile("^(CREATE TABLE|ALTER TABLE) (assistant_[a-z_]+)\\b(.*)", Pattern.DOTALL).matcher(sql);
             if(m.find() && DDL_TABLES.contains(m.group(2)) && (m.group(1).startsWith("CREATE") || m.group(3).strip().startsWith("ADD COLUMN")))
             {
@@ -420,7 +425,8 @@ class AssistantDefaultsChangelogTest extends BaseAssistantTest
 
 
    /*******************************************************************************
-    ** SQL statements, without comment lines, split at a ';' that ends a line.
+    ** SQL statements, without comment lines, split at a ';' that ends a line;
+    ** a DO $$ ... $$ block stays one statement.
     *******************************************************************************/
    static List<String> statements(String sql)
    {
@@ -433,14 +439,36 @@ class AssistantDefaultsChangelogTest extends BaseAssistantTest
             text.append(stripped).append("\n");
          }
       }
-      List<String> rs = new ArrayList<>();
+      List<String>  rs    = new ArrayList<>();
+      StringBuilder block = null;
       for(String statement : text.toString().split(";\\s*\\n"))
       {
+         String stripped = statement.strip();
+         if(block == null && stripped.startsWith("DO $$") && !stripped.endsWith("$$"))
+         {
+            // a dollar-quoted block has its own semicolons: keep it whole up to its closing $$
+            block = new StringBuilder(stripped);
+            continue;
+         }
+         if(block != null)
+         {
+            block.append(";\n").append(stripped);
+            if(stripped.endsWith("$$"))
+            {
+               rs.add(block.toString());
+               block = null;
+            }
+            continue;
+         }
          if(!statement.isBlank())
          {
-            rs.add(statement.strip().replaceAll(";$", ""));
+            rs.add(stripped.replaceAll(";$", ""));
          }
       }
+      if(block != null)
+      {
+         rs.add(block.toString());
+      }
       return (rs);
    }
 

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/TranscriptSchemaTest.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant;
+
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.DriverManager;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Execute the payload column definitions and upgrade statements on H2. Other
+ ** schema constraints have their own Postgres guardrail; this checks expansion.
+ *******************************************************************************/
+class TranscriptSchemaTest
+{
+   /*******************************************************************************
+    ** Both fresh installs and upgraded columns accept expanded token sequences.
+    *******************************************************************************/
+   @Test
+   void redactedColumnsAcceptExpandedTokens() throws Exception
+   {
+      Map<String, List<String>> columns = Map.of(
+         "assistant_turn", List.of("stop_reason"),
+         "assistant_tool_call", List.of("tool_call_id", "result_summary"),
+         "assistant_feedback", List.of("reason"),
+         "assistant_usage", List.of("request_id"),
+         "assistant_pending_action", List.of("tool_call_id", "summary", "result_summary"),
+         "assistant_data_audit", List.of("reason"),
+         "assistant_budget_override", List.of("reason"),
+         "assistant_scheduled_run", List.of("summary", "error"));
+      String ddl = Files.readString(Path.of("..", "docs", "schema-postgres.sql"));
+      String tokens = "hmac:v1:test:0123456789abcdefghijklmnopqrstuvwxyzABCDEFG".repeat(100);
+      try(var connection = DriverManager.getConnection("jdbc:h2:mem:transcript" + System.nanoTime() + ";MODE=PostgreSQL");
+         var statement = connection.createStatement())
+      {
+         for(var table : columns.entrySet())
+         {
+            var create = Pattern.compile("CREATE TABLE " + table.getKey() + "\\s*\\((.*?)\\n\\);", Pattern.DOTALL).matcher(ddl);
+            assertThat(create.find()).as(table.getKey()).isTrue();
+            for(String column : table.getValue())
+            {
+               var definition = Pattern.compile("(?m)^   " + column + " (TEXT|VARCHAR\\(\\d+\\))").matcher(create.group(1));
+               assertThat(definition.find()).as(column).isTrue();
+               statement.execute("DROP TABLE IF EXISTS " + table.getKey());
+               statement.execute("CREATE TABLE " + table.getKey() + " (" + column + " " + definition.group(1) + ")");
+               try(var insert = connection.prepareStatement("INSERT INTO " + table.getKey() + " VALUES (?)"))
+               {
+                  insert.setString(1, tokens);
+                  assertThat(insert.executeUpdate()).isEqualTo(1);
+               }
+               String migration = "ALTER TABLE " + table.getKey() + " ALTER COLUMN " + column + " SET DATA TYPE TEXT";
+               if(!"result_summary".equals(column) && !("assistant_pending_action".equals(table.getKey()) && "summary".equals(column)))
+               {
+                  assertThat(ddl).contains(migration + ";");
+                  statement.execute("DELETE FROM " + table.getKey());
+                  statement.execute("ALTER TABLE " + table.getKey() + " ALTER COLUMN " + column + " SET DATA TYPE VARCHAR(64)");
+                  statement.execute(migration);
+                  statement.execute(migration);
+                  try(var insert = connection.prepareStatement("INSERT INTO " + table.getKey() + " VALUES (?)"))
+                  {
+                     insert.setString(1, tokens);
+                     assertThat(insert.executeUpdate()).isEqualTo(1);
+                  }
+               }
+               try(var result = statement.executeQuery("SELECT " + column + " FROM " + table.getKey()))
+               {
+                  assertThat(result.next()).isTrue();
+                  assertThat(result.getString(1)).isEqualTo(tokens);
+               }
+            }
+         }
+      }
+   }
+}

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/api/AssistantRouteProviderTest.java

@@ -30,9 +30,13 @@ import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.kingsrook.qbits.assistant.model.AssistantPolicy;
 import com.kingsrook.qbits.assistant.persistence.AssistantStore;
+import com.kingsrook.qbits.assistant.spi.ModelPart;
+import com.kingsrook.qbits.assistant.spi.ModelResponse;
 import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
 import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
 import com.kingsrook.qbits.assistant.testing.UiApiTestServer;
@@ -314,7 +318,7 @@ class AssistantRouteProviderTest extends BaseAssistantTest
       assertThat(send("POST", "/messages/" + messages.get(0).get("id") + "/feedback", EDITOR, "{\"rating\":1}").statusCode()).isEqualTo(400);
       assertThat(send("POST", "/messages/" + msgId + "/feedback", OTHER_EDITOR, "{\"rating\":-1}").statusCode()).isEqualTo(404);
 
-      assertThat(send("POST", "/turns/" + turnId + "/cancel", EDITOR, "{}").statusCode()).isEqualTo(409);
+      assertThat(send("POST", "/turns/" + turnId + "/cancel", EDITOR, "{}").statusCode()).isEqualTo(200);
    }
 
 
@@ -351,15 +355,10 @@ class AssistantRouteProviderTest extends BaseAssistantTest
       }
       assertThat(turnId).isNotNull();
 
-      ///////////////////////////////////////////////////////////////
-      // a second turn while one runs: refused (concurrency = 1)  //
-      ///////////////////////////////////////////////////////////////
-      assertThat(send("POST", "/conversations/" + id + "/turns", EDITOR, "{\"text\":\"again\"}").statusCode()).isEqualTo(429);
-
       Thread.sleep(500);
       assertThat(send("POST", "/turns/" + turnId + "/cancel", OTHER_EDITOR, "{}").statusCode()).isEqualTo(404);
       HttpResponse<String> cancel = send("POST", "/turns/" + turnId + "/cancel", EDITOR, "{}");
-      assertThat(cancel.statusCode()).isEqualTo(202);
+      assertThat(cancel.statusCode()).isEqualTo(200);
       assertThat(json(cancel)).containsEntry("status", "CANCELLING");
 
       List<Frame> frames = frames(future.get().body());
@@ -371,6 +370,96 @@ class AssistantRouteProviderTest extends BaseAssistantTest
 
 
 
+   /*******************************************************************************
+    ** A disconnected POST keeps running; active discovery and replay can attach
+    ** again while the answer is blocked, then receive the terminal frame.
+    *******************************************************************************/
+   @Test
+   @SuppressWarnings("unchecked")
+   void testDisconnectRefreshAndReattach() throws Exception
+   {
+      CountDownLatch entered = new CountDownLatch(1);
+      CountDownLatch release = new CountDownLatch(1);
+      provider.thenToolUses("I am checking the count before answering. ", List.<Object[]>of(new Object[] { "count_records", Map.of("table", "blog_post") }))
+         .thenRespond(request ->
+         {
+            entered.countDown();
+            try
+            {
+               if(!release.await(5, TimeUnit.SECONDS))
+               {
+                  throw (new IllegalStateException("test did not release provider"));
+               }
+            }
+            catch(InterruptedException e)
+            {
+               Thread.currentThread().interrupt();
+            }
+            return (new ModelResponse().withParts(List.of(ModelPart.text("Finished after reconnect."))).withStopReason(ModelResponse.STOP_END_TURN));
+         });
+      Integer conversation = (Integer) json(send("POST", "/conversations", EDITOR, "{}")).get("id");
+      HttpRequest request = HttpRequest.newBuilder(URI.create(base + "/conversations/" + conversation + "/turns"))
+         .header("Cookie", "sessionId=" + EDITOR).header("Content-Type", "application/json")
+         .POST(HttpRequest.BodyPublishers.ofString("{\"text\":\"count posts\"}")).build();
+      int lastSeq = 0;
+      try
+      {
+         HttpResponse<InputStream> response = client.send(request, HttpResponse.BodyHandlers.ofInputStream());
+         assertThat(response.statusCode()).isEqualTo(200);
+         try(BufferedReader reader = new BufferedReader(new InputStreamReader(response.body(), StandardCharsets.UTF_8)))
+         {
+            String event = null;
+            String line;
+            while((line = reader.readLine()) != null)
+            {
+               if(line.startsWith("id: "))
+               {
+                  lastSeq = Integer.parseInt(line.substring(4));
+               }
+               if(line.startsWith("event: "))
+               {
+                  event = line.substring(7);
+               }
+               if(line.isEmpty() && "text.delta".equals(event))
+               {
+                  break;
+               }
+            }
+         }
+         assertThat(lastSeq).isPositive();
+         assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue();
+         HttpResponse<String> activeResponse = send("GET", "/turns/active", EDITOR, null);
+         assertThat(activeResponse.statusCode()).isEqualTo(200);
+         List<Map<String, Object>> active = (List<Map<String, Object>>) json(activeResponse).get("turns");
+         assertThat(active).hasSize(1);
+         Map<String, Object> turn = active.getFirst();
+         assertThat(turn).containsEntry("live", true).containsEntry("phase", "thinking").containsEntry("conversationId", conversation)
+            .containsKeys("startedAt", "heartbeatAt", "elapsedMs", "concurrencyMode");
+         assertThat(json(send("GET", "/turns/active", OTHER_EDITOR, null)).get("turns")).isEqualTo(List.of());
+         assertThat(send("GET", "/turns/active", NO_ACCESS, null).statusCode()).isEqualTo(403);
+         HttpRequest attach = HttpRequest.newBuilder(URI.create(base + "/turns/" + turn.get("turnId") + "/events?after=0"))
+            .header("Cookie", "sessionId=" + EDITOR).GET().build();
+         HttpResponse<InputStream> attached = client.send(attach, HttpResponse.BodyHandlers.ofInputStream());
+         release.countDown();
+         List<Frame> followed = frames(attached.body());
+         assertThat(followed.getFirst().event()).isEqualTo("message.start");
+         assertThat(followed.getFirst().data()).containsKey("startedAt");
+         assertThat(followed.getLast().event()).isEqualTo("message.end");
+         assertThat(followed.getLast().data()).containsEntry("status", "COMPLETE");
+         HttpRequest replay = HttpRequest.newBuilder(URI.create(base + "/turns/" + turn.get("turnId") + "/events?after=" + lastSeq))
+            .header("Cookie", "sessionId=" + EDITOR).GET().build();
+         List<Frame> rest = frames(client.send(replay, HttpResponse.BodyHandlers.ofInputStream()).body());
+         assertThat(Integer.parseInt(rest.getFirst().id())).isEqualTo(lastSeq + 1);
+         assertThat(rest.getLast().event()).isEqualTo("message.end");
+      }
+      finally
+      {
+         release.countDown();
+      }
+   }
+
+
+
    /*******************************************************************************
     **
     *******************************************************************************/

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/api/TurnStreamsTest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.api;
+
+
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.assistant.engine.TurnEvent;
+import com.kingsrook.qbits.assistant.engine.TurnEventLog;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Terminal events must be drained even when closure races an empty poll.
+ *******************************************************************************/
+class TurnStreamsTest
+{
+   /*******************************************************************************
+    ** Reproduce the poll/close boundary without relying on thread timing.
+    *******************************************************************************/
+   @Test
+   void drainsTerminalEventsAfterEmptyPoll() throws Exception
+   {
+      TurnEventLog log = new TurnEventLog()
+      {
+         /*******************************************************************************
+          ** Append just after a timed-out poll, before the reader checks closure.
+          *******************************************************************************/
+         @Override
+         public List<TurnEvent> after(int afterSeq, long timeoutMillis) throws InterruptedException
+         {
+            if(size() == 0)
+            {
+               append(TurnEvent.ERROR, Map.of("code", "TIMEOUT"));
+               append(TurnEvent.MESSAGE_END, Map.of("status", "ERROR"));
+               return (List.of());
+            }
+            return (super.after(afterSeq, timeoutMillis));
+         }
+      };
+      ByteArrayOutputStream output = new ByteArrayOutputStream();
+      TurnStreams.writeLive(output, log, 0);
+      assertThat(output.toString(StandardCharsets.UTF_8)).contains("event: error", "TIMEOUT", "event: message.end");
+   }
+}

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/DurableTurnsTest.java

@@ -0,0 +1,985 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.engine;
+
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+import com.kingsrook.qbits.assistant.AssistantQBitConfig;
+import com.kingsrook.qbits.assistant.AssistantRuntime;
+import com.kingsrook.qbits.assistant.api.AssistantApi;
+import com.kingsrook.qbits.assistant.api.AssistantApiException;
+import com.kingsrook.qbits.assistant.jobs.AssistantRunTracker;
+import com.kingsrook.qbits.assistant.model.AssistantConversation;
+import com.kingsrook.qbits.assistant.model.AssistantMessage;
+import com.kingsrook.qbits.assistant.model.AssistantTurn;
+import com.kingsrook.qbits.assistant.persistence.AssistantStore;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
+import com.kingsrook.qbits.assistant.provider.fake.ScriptedModelProvider;
+import com.kingsrook.qbits.assistant.spi.ModelEventListener;
+import com.kingsrook.qbits.assistant.spi.ModelPart;
+import com.kingsrook.qbits.assistant.spi.ModelRequest;
+import com.kingsrook.qbits.assistant.spi.ModelResponse;
+import com.kingsrook.qbits.assistant.spi.ModelUsage;
+import com.kingsrook.qbits.assistant.spi.StreamHandle;
+import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
+import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
+import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.data.QRecordEntity;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static com.kingsrook.qbits.assistant.testing.AssistantTestFixture.EDITOR;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Durable run lifecycle, using latches instead of provider timing guesses.
+ *******************************************************************************/
+class DurableTurnsTest extends BaseAssistantTest
+{
+   private static final String CLASSIFIED_TEXT = "CLASSIFIED-LIFECYCLE-CANARY";
+
+   private QSession session;
+   private AssistantApi api;
+   private AssistantStore store;
+
+
+
+   /*******************************************************************************
+    ** Short, valid lease settings keep watchdog tests bounded.
+    *******************************************************************************/
+   @Override
+   protected void customizeConfig(AssistantQBitConfig config)
+   {
+      config.withTurnHeartbeatSeconds(1).withTurnLeaseSeconds(3).withTurnForceFinalizeGraceSeconds(1)
+         .withMaxTurnSeconds(10).withKillSwitchPollMillis(50);
+   }
+
+
+
+   /*******************************************************************************
+    ** Every operation uses the user's own fixture session.
+    *******************************************************************************/
+   @BeforeEach
+   void atomicBackend()
+   {
+      config.getRuntime().close();
+      config.setRuntime(new AssistantRuntime(qInstance, config, Clock.systemUTC(), new AtomicCallsStore(config)));
+      userContext();
+   }
+
+
+
+   /*******************************************************************************
+    ** Establish the caller's context, also after a test replaces the runtime.
+    *******************************************************************************/
+   private void userContext()
+   {
+      session = AssistantTestFixture.ownerSession(EDITOR, config);
+      QContext.init(qInstance, session);
+      api = new AssistantApi(config.getRuntime(), session);
+      store = config.getRuntime().getStore();
+   }
+
+
+
+   /*******************************************************************************
+    ** Terminal paths on other threads flush the same provisional transcript scope.
+    ** Live text stays intact; stored user/message/replay content has only tokens.
+    *******************************************************************************/
+   @ParameterizedTest
+   @ValueSource(strings = { "SUPERSEDED", "TIMEOUT", "SHUTDOWN" })
+   void terminalPathsRedactClassifiedContent(String reason) throws Exception
+   {
+      config.withDataClassification("blog_post", "title", DataClassification.PII);
+      ControlledProvider controlled = controlled(true);
+      controlled.classifiedText = CLASSIFIED_TEXT;
+      if("TIMEOUT".equals(reason))
+      {
+         config.setMaxTurnSeconds(1);
+      }
+      Integer conversation = conversation();
+      RunningTurn run = api.startTurn(conversation, Map.of("text", CLASSIFIED_TEXT));
+      try
+      {
+         controlled.awaitCalls(1);
+         if("SHUTDOWN".equals(reason))
+         {
+            config.getRuntime().shutdown(Duration.ZERO);
+            QContext.init(qInstance, session);
+         }
+         else if("SUPERSEDED".equals(reason))
+         {
+            api.startTurn(conversation, Map.of("text", "next question"));
+         }
+         AssistantTurn row = completed(run);
+         assertThat(row.getCancelReason()).isEqualTo(reason);
+         assertThat(store.getMessage(EDITOR, row.getUserMessageId()).getPartsJson()).contains("changed:v1").doesNotContain(CLASSIFIED_TEXT);
+         assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains("changed:v1").doesNotContain(CLASSIFIED_TEXT);
+         assertThat(api.turnEvents(run.getTurnId(), 0).persisted()).anySatisfy(event -> assertThat(event.toJson()).contains("changed:v1"))
+            .allSatisfy(event -> assertThat(event.toJson()).doesNotContain(CLASSIFIED_TEXT));
+         assertThat(run.getLog().after(0, 1)).anySatisfy(event -> assertThat(event.toJson()).contains(CLASSIFIED_TEXT));
+         assertThat(TranscriptRedactor.current()).isNull();
+         assertThat(QLogger.redact(CLASSIFIED_TEXT)).isEqualTo("changed:v1");
+      }
+      finally
+      {
+         controlled.release();
+      }
+      await(() -> CLASSIFIED_TEXT.equals(QLogger.redact(CLASSIFIED_TEXT)));
+   }
+
+
+
+   /*******************************************************************************
+    ** A rejected submission still persists its terminal notice and releases scope.
+    *******************************************************************************/
+   @Test
+   void rejectedSubmissionFinalizesTranscript() throws Exception
+   {
+      Integer conversation = conversation();
+      config.getRuntime().getExecutor().shutdown();
+      assertThatThrownBy(() -> api.startTurn(conversation, Map.of("text", "Submission failed")))
+         .isInstanceOf(java.util.concurrent.RejectedExecutionException.class);
+      QContext.init(qInstance, session);
+      assertThat(store.listRunningTurns(EDITOR)).isEmpty();
+      assertThat(store.listMessages(EDITOR, conversation, null, 100)).anySatisfy(message ->
+         assertThat(message.getPartsJson()).contains(TurnEngine.RESTART_TEXT));
+      assertThat(TranscriptRedactor.current()).isNull();
+   }
+
+
+
+   /*******************************************************************************
+    ** Completion closes the lease, emits progress, and frees the next question.
+    *******************************************************************************/
+   @Test
+   void completion() throws Exception
+   {
+      provider.thenText("Done").thenText("Again");
+      Integer conversation = conversation();
+      RunningTurn first = api.startTurn(conversation, Map.of("text", "hello"));
+      AssistantTurn row = completed(first);
+      assertThat(row.getStatus()).isEqualTo("COMPLETE");
+      assertThat(row.getPhase()).isEqualTo("done");
+      assertThat(row.getEndedAt()).isNotNull();
+      assertThat(row.getHeartbeatAt()).isAfterOrEqualTo(row.getStartedAt());
+      assertThat(row.getLeaseOwner()).isEqualTo(config.getRuntime().getInstanceId()).hasSizeLessThanOrEqualTo(100);
+      assertThat(row.getConcurrencyMode()).isEqualTo("SUPERSEDE");
+      assertThat(store.listRunningTurns(EDITOR)).isEmpty();
+      assertThat(store.updateRunningTurn(EDITOR, new QRecord().withValue("id", first.getTurnId()).withValue("status", "CANCELLING"))).isFalse();
+      assertThat(first.getLog().after(0, 1).getFirst().getData()).containsEntry("startedAt", row.getStartedAt().toString());
+      assertThat(first.getLog().after(0, 1)).filteredOn(e -> TurnEvent.TURN_PHASE.equals(e.getType()))
+         .extracting(e -> e.getData().get("phase")).containsExactly("thinking", "writing");
+      assertThat(completed(api.startTurn(conversation, Map.of("text", "again"))).getStatus()).isEqualTo("COMPLETE");
+   }
+
+
+
+   /*******************************************************************************
+    ** A throwing post-processing collaborator cannot strand a streamed answer.
+    *******************************************************************************/
+   @Test
+   void accountingFailureStillFinalizes() throws Exception
+   {
+      AssistantStore failing = new AssistantStore(config)
+      {
+         /*******************************************************************************
+          ** Inject a failure after streaming has finished.
+          *******************************************************************************/
+         @Override
+         public void addUsage(String userId, LocalDate date, ModelUsage usage, long costMicros, boolean countTurn)
+         {
+            throw (new IllegalStateException("accounting unavailable"));
+         }
+
+         /*******************************************************************************
+          ** Conversation totals fail independently of the turn update.
+          *******************************************************************************/
+         @Override
+         public void updateFields(String table, QRecord record) throws QException
+         {
+            if(AssistantConversation.TABLE_NAME.equals(table) && record.getValue("totalTokensIn") != null)
+            {
+               throw (new IllegalStateException("totals unavailable"));
+            }
+            super.updateFields(table, record);
+         }
+      };
+      config.getRuntime().close();
+      AssistantRuntime runtime = new AssistantRuntime(qInstance, config, Clock.systemUTC())
+      {
+         /*******************************************************************************
+          ** Substitute only the engine/API persistence collaborator.
+          *******************************************************************************/
+         @Override
+         public AssistantStore getStore()
+         {
+            return (failing);
+         }
+      };
+      config.setRuntime(runtime);
+      userContext();
+      provider.thenText("A streamed answer").thenText("Next answer");
+      Integer conversation = conversation();
+      RunningTurn run = api.startTurn(conversation, Map.of("text", "hello"));
+      assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
+      assertThat(api.turnEvents(run.getTurnId(), 0).persisted()).extracting(TurnEvent::getType).contains(TurnEvent.TEXT_DELTA, TurnEvent.MESSAGE_END);
+      assertThat(completed(api.startTurn(conversation, Map.of("text", "next"))).getStatus()).isEqualTo("COMPLETE");
+   }
+
+
+
+   /*******************************************************************************
+    ** Explicit concurrent mode permits overlap within and across conversations.
+    *******************************************************************************/
+   @Test
+   void concurrentQuestions() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      Integer firstConversation = conversation();
+      List<RunningTurn> runs = new ArrayList<>();
+      runs.add(api.startTurn(firstConversation, Map.of("text", "one", "concurrency", "CoNcUrReNt")));
+      controlled.awaitCalls(1);
+      runs.add(api.startTurn(firstConversation, Map.of("text", "two", "concurrency", "concurrent")));
+      controlled.awaitCalls(2);
+      runs.add(api.startTurn(conversation(), Map.of("text", "three", "concurrency", "concurrent")));
+      controlled.awaitCalls(3);
+      assertThat(active()).hasSize(3);
+      controlled.release();
+      for(RunningTurn run : runs)
+      {
+         assertThat(completed(run).getStatus()).isEqualTo("COMPLETE");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Default supersede frees the slot immediately and records the explanation.
+    *******************************************************************************/
+   @Test
+   void supersedeAndSoftCap() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      Integer conversation = conversation();
+      RunningTurn first = api.startTurn(conversation, Map.of("text", "one"));
+      controlled.awaitCalls(1);
+      RunningTurn second = api.startTurn(conversation, Map.of("text", "two"));
+      controlled.awaitCalls(2);
+      AssistantTurn superseded = completed(first);
+      assertThat(superseded.getStatus()).isEqualTo("CANCELLED");
+      assertThat(superseded.getCancelReason()).isEqualTo("SUPERSEDED");
+      assertThat(store.getMessage(EDITOR, superseded.getAssistantMessageId()).getPartsJson()).contains(TurnEngine.SUPERSEDED_TEXT);
+      config.setMaxConcurrentTurnsPerUser(2);
+      RunningTurn third = api.startTurn(conversation(), Map.of("text", "three", "concurrency", "concurrent"));
+      controlled.awaitCalls(3);
+      RunningTurn fourth = api.startTurn(conversation(), Map.of("text", "four", "concurrency", "concurrent"));
+      controlled.awaitCalls(4);
+      assertThat(completed(second).getCancelReason()).isEqualTo("SUPERSEDED");
+      controlled.release();
+      assertThat(completed(third).getStatus()).isEqualTo("COMPLETE");
+      assertThat(completed(fourth).getStatus()).isEqualTo("COMPLETE");
+   }
+
+
+
+   /*******************************************************************************
+    ** A provider ignoring both cancel and interruption is forcibly finalized.
+    *******************************************************************************/
+   @Test
+   void wedgedProviderTimeout() throws Exception
+   {
+      ControlledProvider controlled = controlled(true);
+      controlled.emitText = false;
+      RecordingTracker tracker = new RecordingTracker();
+      config.setRunTracker(tracker);
+      config.setMaxTurnSeconds(1);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wedged"));
+      try
+      {
+         controlled.awaitCalls(1);
+         AssistantTurn row = completed(run);
+         assertThat(row.getStatus()).isEqualTo("ERROR");
+         assertThat(row.getErrorCode()).isEqualTo("TIMEOUT");
+         assertThat(row.getCancelReason()).isEqualTo("TIMEOUT");
+         assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains(TurnEngine.TIMEOUT_TEXT);
+         assertThat(controlled.handles.getFirst().cancelled).isTrue();
+         assertThat(tracker.finishes.get()).isEqualTo(1);
+         int events = run.getLog().after(0, 1).size();
+         controlled.release();
+         await(() -> controlled.handles.getFirst().returned);
+         assertThat(run.getLog().after(0, 1)).hasSize(events);
+         assertThat(tracker.finishes.get()).isEqualTo(1);
+      }
+      finally
+      {
+         controlled.release();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Stop has its own bounded force path even when the provider ignores it.
+    *******************************************************************************/
+   @Test
+   void stopWedgedProvider() throws Exception
+   {
+      ControlledProvider controlled = controlled(true);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      try
+      {
+         controlled.awaitCalls(1);
+         assertThat(api.cancelTurn(run.getTurnId())).containsEntry("status", "CANCELLING");
+         AssistantTurn row = completed(run);
+         assertThat(row.getStatus()).isEqualTo("CANCELLED");
+         assertThat(row.getCancelReason()).isEqualTo("USER");
+      }
+      finally
+      {
+         controlled.release();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A failed heartbeat tick does not stop cancellation polling on later ticks.
+    *******************************************************************************/
+   @Test
+   void heartbeatFailureDoesNotKillWatchdog() throws Exception
+   {
+      AtomicInteger heartbeats = new AtomicInteger();
+      AssistantStore failingOnce = new AssistantStore(config)
+      {
+         /*******************************************************************************
+          ** Fail only the first renewal, then behave like the real store.
+          *******************************************************************************/
+         @Override
+         public void updateFields(String table, QRecord record) throws QException
+         {
+            if(AssistantTurn.TABLE_NAME.equals(table) && record.getValue("heartbeatAt") != null && heartbeats.incrementAndGet() == 1)
+            {
+               throw (new QException("temporary heartbeat failure"));
+            }
+            super.updateFields(table, record);
+         }
+      };
+      config.getRuntime().close();
+      config.setRuntime(new AssistantRuntime(qInstance, config)
+      {
+         /*******************************************************************************
+          ** A transient persistence failure local to the owner.
+          *******************************************************************************/
+         @Override
+         public AssistantStore getStore()
+         {
+            return (failingOnce);
+         }
+      });
+      userContext();
+      ControlledProvider controlled = controlled(false);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      controlled.awaitCalls(1);
+      AssistantRuntime other = new AssistantRuntime(qInstance, config);
+      try
+      {
+         new AssistantApi(other, session).cancelTurn(run.getTurnId());
+         assertThat(completed(run).getStatus()).isEqualTo("CANCELLED");
+         assertThat(heartbeats.get()).isGreaterThanOrEqualTo(2);
+      }
+      finally
+      {
+         controlled.release();
+         other.close();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Stale leases, old cancel requests and absolute deadlines release remotely.
+    *******************************************************************************/
+   @Test
+   void orphanRecoveryAndIdempotentStop() throws Exception
+   {
+      AssistantTurn stale = remote(Instant.now().minusSeconds(5));
+      store.updateFields(AssistantMessage.TABLE_NAME, new QRecord().withValue("id", stale.getAssistantMessageId())
+         .withValue("partsJson", new MessageParts().addText("changed:v1").toJson()));
+      AssistantTurn fresh = remote(Instant.now());
+      AssistantTurn deadline = remote(Instant.now());
+      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", deadline.getId()).withValue("startedAt", Instant.now().minusSeconds(30)));
+      assertThat(active()).extracting(r -> r.get("turnId")).containsExactly(fresh.getId());
+      assertThat(store.getTurn(EDITOR, stale.getId()).getStatus()).isEqualTo("ERROR");
+      assertThat(store.getMessage(EDITOR, stale.getAssistantMessageId()).getPartsJson()).contains("changed:v1", "server restarted");
+      assertThat(store.getTurn(EDITOR, deadline.getId()).getStatus()).isEqualTo("ERROR");
+      assertThat(api.turnEvents(stale.getId(), 0).persisted()).extracting(TurnEvent::getType).containsExactly(TurnEvent.ERROR, TurnEvent.MESSAGE_END);
+      assertThat(api.turnEvents(stale.getId(), 0).persisted()).hasSize(2);
+      assertThat(api.cancelTurn(stale.getId())).containsEntry("status", "ERROR");
+      assertThat(api.cancelTurn(fresh.getId())).containsEntry("status", "CANCELLING");
+      assertThat(store.getTurn(EDITOR, fresh.getId()).getCancelRequestedAt()).isNotNull();
+      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", fresh.getId()).withValue("cancelRequestedAt", Instant.now().minusSeconds(5)));
+      assertThat(active()).isEmpty();
+      assertThat(store.getTurn(EDITOR, fresh.getId()).getStatus()).isEqualTo("CANCELLED");
+      AssistantTurn stopped = remote(Instant.now().minusSeconds(5));
+      assertThat(api.cancelTurn(stopped.getId())).containsEntry("status", "CANCELLED");
+      assertThat(api.cancelTurn(stopped.getId())).containsEntry("status", "CANCELLED");
+      provider.thenText("Recovered");
+      assertThat(completed(api.startTurn(conversation(), Map.of("text", "next"))).getStatus()).isEqualTo("COMPLETE");
+   }
+
+
+
+   /*******************************************************************************
+    ** Two replicas reaping the same row produce only one terminal frame.
+    *******************************************************************************/
+   @Test
+   void simultaneousReapers() throws Exception
+   {
+      AssistantTurn stale = remote(Instant.now().minusSeconds(5));
+      AssistantRuntime other = new AssistantRuntime(qInstance, config);
+      CountDownLatch start = new CountDownLatch(1);
+      try(var executor = java.util.concurrent.Executors.newVirtualThreadPerTaskExecutor())
+      {
+         List<java.util.concurrent.Future<?>> futures = new ArrayList<>();
+         for(AssistantRuntime runtime : List.of(config.getRuntime(), other))
+         {
+            futures.add(executor.submit(() ->
+            {
+               start.await();
+               QContext.init(qInstance, session);
+               try
+               {
+                  new AssistantApi(runtime, session).activeTurns();
+               }
+               finally
+               {
+                  QContext.clear();
+               }
+               return (null);
+            }));
+         }
+         start.countDown();
+         for(var future : futures)
+         {
+            future.get(5, java.util.concurrent.TimeUnit.SECONDS);
+         }
+         List<TurnEvent> events = api.turnEvents(stale.getId(), 0).persisted();
+         assertThat(events).filteredOn(TurnEvent::isTerminal).hasSize(1);
+         assertThat(store.getTurn(EDITOR, stale.getId()).getStatus()).isEqualTo("ERROR");
+      }
+      finally
+      {
+         other.close();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A terminal outcome written elsewhere wins over the local engine's finish.
+    *******************************************************************************/
+   @Test
+   void remoteTerminalOutcomeIsPreserved() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      controlled.awaitCalls(1);
+      store.insertEvent(run.getTurnId(), EDITOR, 50, TurnEvent.MESSAGE_END,
+         new TurnEvent(50, TurnEvent.MESSAGE_END, Map.of("status", "ERROR", "stopReason", "interrupted")).toJson());
+      store.updateFields(AssistantTurn.TABLE_NAME, new QRecord().withValue("id", run.getTurnId()).withValue("status", "ERROR")
+         .withValue("phase", "done").withValue("errorCode", "INTERNAL").withValue("stopReason", "interrupted").withValue("endedAt", Instant.now()));
+      controlled.release();
+      assertThat(completed(run).getStatus()).isEqualTo("ERROR");
+      assertThat(api.turnEvents(run.getTurnId(), 0).persisted()).filteredOn(TurnEvent::isTerminal).hasSize(1);
+      assertThat(run.getLog().after(0, 1)).last().extracting(e -> e.getData().get("status")).isEqualTo("ERROR");
+   }
+
+
+
+   /*******************************************************************************
+    ** Configuration rejects invalid lease timing and concurrency modes.
+    *******************************************************************************/
+   @Test
+   void configurationValidation()
+   {
+      config.withTurnHeartbeatSeconds(5).withTurnLeaseSeconds(4).withDefaultConcurrencyMode("invalid");
+      List<String> errors = new ArrayList<>();
+      config.validate(qInstance, errors);
+      assertThat(errors).contains("turnHeartbeatSeconds must be less than turnLeaseSeconds.", "defaultConcurrencyMode must be SUPERSEDE or CONCURRENT.");
+      config.withRunTracker(null);
+      assertThat(config.getRunTracker()).isSameAs(AssistantRunTracker.NOOP);
+   }
+
+
+
+   /*******************************************************************************
+    ** A different runtime writes Stop; the owner sees it within one heartbeat.
+    *******************************************************************************/
+   @Test
+   void crossPodStopAndHeartbeat() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      controlled.awaitCalls(1);
+      AssistantRuntime remote = new AssistantRuntime(qInstance, config);
+      try
+      {
+         assertThat(remote.getInstanceId()).isNotEqualTo(config.getRuntime().getInstanceId());
+         AssistantApi otherPod = new AssistantApi(remote, session);
+         assertThat(((List<?>) otherPod.activeTurns().get("turns"))).hasSize(1);
+         assertThat(active().getFirst()).containsEntry("live", true).containsEntry("phase", "writing").containsKey("elapsedMs");
+         assertThat(otherPod.cancelTurn(run.getTurnId())).containsEntry("status", "CANCELLING");
+         AssistantTurn row = completed(run);
+         assertThat(row.getStatus()).isEqualTo("CANCELLED");
+         assertThat(row.getCancelReason()).isEqualTo("USER");
+         assertThat(row.getHeartbeatAt()).isAfter(row.getStartedAt());
+         assertThat(otherPod.cancelTurn(run.getTurnId())).containsEntry("status", "CANCELLED");
+      }
+      finally
+      {
+         controlled.release();
+         remote.close();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Shutdown force-finalizes a wedged engine and refuses new inserts with 503.
+    *******************************************************************************/
+   @Test
+   void rollingDeploy() throws Exception
+   {
+      ControlledProvider controlled = controlled(true);
+      RecordingTracker tracker = new RecordingTracker();
+      config.setRunTracker(tracker);
+      Integer conversation = conversation();
+      RunningTurn run = api.startTurn(conversation, Map.of("text", "wait"));
+      try
+      {
+         controlled.awaitCalls(1);
+         config.getRuntime().shutdown(Duration.ZERO);
+         QContext.init(qInstance, session);
+         AssistantTurn row = completed(run);
+         assertThat(row.getStatus()).isEqualTo("ERROR");
+         assertThat(row.getErrorCode()).isEqualTo("INTERRUPTED");
+         assertThat(store.getMessage(EDITOR, row.getAssistantMessageId()).getPartsJson()).contains(TurnEngine.RESTART_TEXT);
+         assertThat(tracker.finishes.get()).isEqualTo(1);
+         int messages = store.listMessages(EDITOR, conversation, null, 100).size();
+         assertThatThrownBy(() -> api.startTurn(conversation, Map.of("text", "retry")))
+            .isInstanceOfSatisfying(AssistantApiException.class, e ->
+            {
+               assertThat(e.getStatus()).isEqualTo(503);
+               assertThat(e.getCode()).isEqualTo("SHUTTING_DOWN");
+            });
+         assertThat(store.listMessages(EDITOR, conversation, null, 100)).hasSize(messages);
+      }
+      finally
+      {
+         controlled.release();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Graceful cancellation also uses the restart outcome, not a user cancel.
+    *******************************************************************************/
+   @Test
+   void gracefulShutdown() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      controlled.awaitCalls(1);
+      config.getRuntime().shutdown(Duration.ofSeconds(1));
+      QContext.init(qInstance, session);
+      assertThat(completed(run).getErrorCode()).isEqualTo("INTERRUPTED");
+   }
+
+
+
+   /*******************************************************************************
+    ** Tracker progress is phase-based; its Stop callback cancels the local run.
+    *******************************************************************************/
+   @Test
+   void trackerAndTrackerFailures() throws Exception
+   {
+      ControlledProvider controlled = controlled(false);
+      RecordingTracker tracker = new RecordingTracker();
+      config.setRunTracker(tracker);
+      RunningTurn run = api.startTurn(conversation(), Map.of("text", "wait"));
+      controlled.awaitCalls(1);
+      await(() -> tracker.cancel != null);
+      tracker.cancel.run();
+      assertThat(completed(run).getStatus()).isEqualTo("CANCELLED");
+      assertThat(tracker.starts.get()).isEqualTo(1);
+      assertThat(tracker.phases).containsExactly("thinking", "writing");
+      assertThat(tracker.finishes.get()).isEqualTo(1);
+      assertThat(tracker.label).isEqualTo("Assistant answer");
+      tracker.fail = true;
+      config.withModelProvider(provider);
+      config.getRuntime().getProviderService().invalidate();
+      provider.thenText("Still works");
+      RunningTurn throwing = api.startTurn(conversation(), Map.of("text", "hello"));
+      assertThat(completed(throwing).getStatus()).isEqualTo("COMPLETE");
+      assertThat(tracker.finishes.get()).isEqualTo(2);
+   }
+
+
+
+   /*******************************************************************************
+    ** Build a remote persisted turn with an assistant message to recover.
+    *******************************************************************************/
+   private AssistantTurn remote(Instant heartbeat) throws Exception
+   {
+      Integer conversation = conversation();
+      AssistantMessage message = store.insert(new AssistantMessage().withConversationId(conversation).withUserId(EDITOR)
+         .withRole("ASSISTANT").withStatus("STREAMING").withSeq(1).withPartsJson("[]"));
+      return (store.insert(new AssistantTurn().withConversationId(conversation).withUserId(EDITOR).withAssistantMessageId(message.getId())
+         .withStatus("RUNNING").withStartedAt(heartbeat).withHeartbeatAt(heartbeat).withLeaseOwner("dead-pod").withPhase("thinking")));
+   }
+
+
+
+   /*******************************************************************************
+    ** Create a user-owned conversation.
+    *******************************************************************************/
+   private Integer conversation() throws Exception
+   {
+      return ((Integer) api.createConversation(Map.of()).get("id"));
+   }
+
+
+
+   /*******************************************************************************
+    ** Read the active-run response.
+    *******************************************************************************/
+   @SuppressWarnings("unchecked")
+   private List<Map<String, Object>> active() throws Exception
+   {
+      return ((List<Map<String, Object>>) api.activeTurns().get("turns"));
+   }
+
+
+
+   /*******************************************************************************
+    ** Wait for both persisted completion and local cleanup, then check replay.
+    *******************************************************************************/
+   private AssistantTurn completed(RunningTurn run) throws Exception
+   {
+      await(() -> config.getRuntime().getRunningTurn(run.getTurnId()) == null);
+      assertThat(run.getLog().isClosed()).isTrue();
+      assertThat(run.getLog().after(0, 1)).filteredOn(TurnEvent::isTerminal).hasSize(1);
+      AssistantTurn row = store.getTurn(EDITOR, run.getTurnId());
+      assertThat(row.getStatus()).isNotIn("RUNNING", "CANCELLING");
+      assertThat(api.turnEvents(run.getTurnId(), 0).persisted()).last().extracting(TurnEvent::getType).isEqualTo(TurnEvent.MESSAGE_END);
+      return (row);
+   }
+
+
+
+   /*******************************************************************************
+    ** Install a deterministic provider before the first call resolves it.
+    *******************************************************************************/
+   private ControlledProvider controlled(boolean stubborn)
+   {
+      ControlledProvider result = new ControlledProvider(stubborn);
+      config.withModelProvider(result);
+      return (result);
+   }
+
+
+
+   /*******************************************************************************
+    ** Short polling with a hard bound, without timing-sensitive fixed waits.
+    *******************************************************************************/
+   private static void await(BooleanSupplier condition) throws InterruptedException
+   {
+      long until = System.nanoTime() + Duration.ofSeconds(6).toNanos();
+      while(!condition.getAsBoolean() && System.nanoTime() < until)
+      {
+         Thread.sleep(5);
+      }
+      assertThat(condition.getAsBoolean()).as("condition reached within six seconds").isTrue();
+   }
+
+
+
+   /*******************************************************************************
+    ** MemoryRecordStore iterates its synchronized maps without holding their
+    ** lock. Serialize individual backend calls, as in RobustnessTest, so the
+    ** fixture models atomic database operations while turns still overlap.
+    *******************************************************************************/
+   private static class AtomicCallsStore extends AssistantStore
+   {
+      private static final Object BACKEND_LOCK = new Object();
+
+      /*******************************************************************************
+       ** Preserve encryption and other configured persistence behavior.
+       *******************************************************************************/
+      AtomicCallsStore(AssistantQBitConfig config)
+      {
+         super(config);
+      }
+
+      /*******************************************************************************
+       ** Make a memory query atomic with respect to insert/update operations.
+       *******************************************************************************/
+      @Override
+      public <T extends QRecordEntity> List<T> query(String table, QQueryFilter filter, Class<T> type) throws QException
+      {
+         synchronized(BACKEND_LOCK)
+         {
+            return (super.query(table, filter, type));
+         }
+      }
+
+      /*******************************************************************************
+       ** Make an insert atomic with respect to memory queries.
+       *******************************************************************************/
+      @Override
+      public <T extends QRecordEntity> T insert(T entity) throws QException
+      {
+         synchronized(BACKEND_LOCK)
+         {
+            return (super.insert(entity));
+         }
+      }
+
+      /*******************************************************************************
+       ** Make an update atomic with respect to memory queries.
+       *******************************************************************************/
+      @Override
+      public void updateFields(String table, QRecord record) throws QException
+      {
+         synchronized(BACKEND_LOCK)
+         {
+            super.updateFields(table, record);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Streams partial text, then blocks until explicitly released.
+    *******************************************************************************/
+   private static class ControlledProvider extends ScriptedModelProvider
+   {
+      private final boolean stubborn;
+      private String classifiedText;
+      private final List<ControlledHandle> handles = new CopyOnWriteArrayList<>();
+      private boolean emitText = true;
+
+      /*******************************************************************************
+       ** Configure whether cancel and interruption are deliberately ignored.
+       *******************************************************************************/
+      ControlledProvider(boolean stubborn)
+      {
+         this.stubborn = stubborn;
+      }
+
+      /*******************************************************************************
+       ** Start a controlled stream.
+       *******************************************************************************/
+      @Override
+      public StreamHandle stream(ModelRequest request, ModelEventListener listener)
+      {
+         ControlledHandle handle = new ControlledHandle(stubborn);
+         if(classifiedText != null)
+         {
+            TranscriptRedactor.current().observe("blog_post", "title", 1, classifiedText, classifiedText);
+            listener.onTextDelta(classifiedText);
+         }
+         if(emitText)
+         {
+            listener.onTextDelta("A partial answer that has already begun streaming to the user. ");
+         }
+         handles.add(handle);
+         return (handle);
+      }
+
+      /*******************************************************************************
+       ** Await a known number of started calls.
+       *******************************************************************************/
+      void awaitCalls(int count) throws InterruptedException
+      {
+         await(() -> handles.size() >= count);
+      }
+
+      /*******************************************************************************
+       ** Always release test threads, including deliberately wedged ones.
+       *******************************************************************************/
+      void release()
+      {
+         handles.forEach(h -> h.release.countDown());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A model that can deliberately violate the provider timeout contract.
+    *******************************************************************************/
+   private static class ControlledHandle implements StreamHandle
+   {
+      private final boolean stubborn;
+      private final CountDownLatch release = new CountDownLatch(1);
+      private volatile boolean cancelled;
+      private volatile boolean returned;
+
+      /*******************************************************************************
+       ** Configure cancellation behavior.
+       *******************************************************************************/
+      ControlledHandle(boolean stubborn)
+      {
+         this.stubborn = stubborn;
+      }
+
+      /*******************************************************************************
+       ** Abort unless this is the wedged-provider test.
+       *******************************************************************************/
+      @Override
+      public void cancel()
+      {
+         cancelled = true;
+         if(!stubborn)
+         {
+            release.countDown();
+         }
+      }
+
+      /*******************************************************************************
+       ** Observe cancellation.
+       *******************************************************************************/
+      @Override
+      public boolean isCancelled()
+      {
+         return (cancelled);
+      }
+
+      /*******************************************************************************
+       ** Block deterministically; stubborn mode ignores timeouts and interrupts.
+       *******************************************************************************/
+      @Override
+      public ModelResponse await(Duration timeout) throws InterruptedException
+      {
+         while(release.getCount() != 0)
+         {
+            try
+            {
+               release.await();
+            }
+            catch(InterruptedException e)
+            {
+               if(!stubborn)
+               {
+                  throw (e);
+               }
+            }
+         }
+         returned = true;
+         return (new ModelResponse().withParts(List.of(ModelPart.text("Done")))
+            .withStopReason(cancelled ? ModelResponse.STOP_CANCELLED : ModelResponse.STOP_END_TURN));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Records the host job-tray contract and can throw from every callback.
+    *******************************************************************************/
+   private static class RecordingTracker implements AssistantRunTracker
+   {
+      private final AtomicInteger starts = new AtomicInteger();
+      private final AtomicInteger finishes = new AtomicInteger();
+      private final List<String> phases = new CopyOnWriteArrayList<>();
+      private volatile Runnable cancel;
+      private String label;
+      private boolean fail;
+
+      /*******************************************************************************
+       ** Record publication.
+       *******************************************************************************/
+      @Override
+      public String start(String userId, Integer turnId, Integer conversationId, String label)
+      {
+         starts.incrementAndGet();
+         this.label = label;
+         if(fail)
+         {
+            throw (new IllegalStateException("tracker start failed"));
+         }
+         return ("job-" + turnId);
+      }
+
+      /*******************************************************************************
+       ** Record a phase transition.
+       *******************************************************************************/
+      @Override
+      public void progress(String trackingId, String phase, String message)
+      {
+         phases.add(phase);
+         if(fail)
+         {
+            throw (new IllegalStateException("tracker progress failed"));
+         }
+      }
+
+      /*******************************************************************************
+       ** Record terminal notification exactly once.
+       *******************************************************************************/
+      @Override
+      public void finish(String trackingId, String status, String message)
+      {
+         finishes.incrementAndGet();
+         if(fail)
+         {
+            throw (new IllegalStateException("tracker finish failed"));
+         }
+      }
+
+      /*******************************************************************************
+       ** Retain the host's Stop callback.
+       *******************************************************************************/
+      @Override
+      public void onCancel(String trackingId, Runnable cancel)
+      {
+         this.cancel = cancel;
+      }
+   }
+}

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/EngineComponentsTest.java

@@ -151,8 +151,7 @@ class EngineComponentsTest extends BaseAssistantTest
 
       RateLimiter limiter = new RateLimiter(clock);
       assertThat(limiter.tryAcquire("u", 2, 1)).isNull();
-      assertThat(limiter.tryAcquire("u", 2, 1)).contains("in progress");
-      limiter.release("u");
+
       assertThat(limiter.tryAcquire("u", 2, 1)).isNull();
       limiter.release("u");
       assertThat(limiter.tryAcquire("u", 2, 1)).contains("last minute");

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/PhiRoutingTest.java

@@ -133,7 +133,7 @@ class PhiRoutingTest extends BaseAssistantTest
       assertThat(outcome.toolCalls()).singleElement().satisfies(call ->
       {
          assertThat(call.getOutcome()).isEqualTo("OK");
-         assertThat(call.getResultJson()).contains("PHI_DIAGNOSIS_SENTINEL");
+         assertThat(call.getResultJson()).doesNotContain("PHI_DIAGNOSIS_SENTINEL").contains("changed:v1");
       });
       assertThat(provider.getRequests()).hasSize(1);
       assertThat(audits()).singleElement().satisfies(row ->

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/RobustnessTest.java

@@ -119,7 +119,7 @@ class RobustnessTest extends BaseAssistantTest
          // idempotent: a second look adds nothing             //
          ////////////////////////////////////////////////////////
          assertThat(api.turnEvents(turn.getId(), 0).persisted()).hasSize(4);
-         assertThatThrownBy(() -> api.cancelTurn(turn.getId())).isInstanceOfSatisfying(AssistantApiException.class, e -> assertThat(e.getStatus()).isEqualTo(409));
+         assertThat(api.cancelTurn(turn.getId())).containsEntry("status", "ERROR");
 
          assertThat(api.turnEvents(young.getId(), 0).finished()).isFalse();
          assertThatThrownBy(() -> api.deleteConversation(conversation.getId())).isInstanceOfSatisfying(AssistantApiException.class, e -> assertThat(e.getStatus()).isEqualTo(409));

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/engine/TranscriptRedactionTest.java

@@ -0,0 +1,476 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2026.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.engine;
+
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import com.kingsrook.qbits.assistant.AssistantQBitConfig;
+import com.kingsrook.qbits.assistant.api.AssistantApi;
+import com.kingsrook.qbits.assistant.classification.ConfiguredClassificationSource;
+import com.kingsrook.qbits.assistant.model.AssistantDataAccess;
+import com.kingsrook.qbits.assistant.model.AssistantDataAudit;
+import com.kingsrook.qbits.assistant.model.AssistantPendingAction;
+import com.kingsrook.qbits.assistant.model.AssistantProvider;
+import com.kingsrook.qbits.assistant.model.AssistantScheduledRun;
+import com.kingsrook.qbits.assistant.model.AssistantToolCall;
+import com.kingsrook.qbits.assistant.model.AssistantTurn;
+import com.kingsrook.qbits.assistant.model.AssistantTurnEvent;
+import com.kingsrook.qbits.assistant.model.AssistantUsage;
+import com.kingsrook.qbits.assistant.persistence.TranscriptRedactor;
+import com.kingsrook.qbits.assistant.spi.AssistantModelProvider;
+import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
+import com.kingsrook.qbits.assistant.spi.ModelPart;
+import com.kingsrook.qbits.assistant.spi.ModelProviderException;
+import com.kingsrook.qbits.assistant.spi.ModelResponse;
+import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
+import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
+import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
+import com.kingsrook.qqq.backend.core.actions.ai.AiTableMetaData;
+import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
+import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueKeySource;
+import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueKeySourceMetaData;
+import com.kingsrook.qqq.backend.core.actions.permissions.fieldaccess.ClassifiedValueTokens;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.audits.AuditLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.audits.QAuditRules;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.layout.PatternLayout;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Secured-data egress canaries: inspect actual backend rows, core audits and logs,
+ ** while independently proving that the eligible model still received the values.
+ *******************************************************************************/
+@Tag("redteam")
+class TranscriptRedactionTest extends BaseAssistantTest
+{
+   private final String phi = "CANARY-PHI-" + UUID.randomUUID();
+   private final String pii = "CANARY-PII-" + UUID.randomUUID();
+
+
+
+   /*******************************************************************************
+    ** Enable the fixture table and a local eligible provider.
+    *******************************************************************************/
+   @Override
+   protected void customizeConfig(AssistantQBitConfig config)
+   {
+      config.withDataClassification("patient", DataClassification.PHI).withProviderType(new PatientProviderType());
+   }
+
+
+
+   /*******************************************************************************
+    ** Real core metadata, including table fallback, drives the redactor and core audits.
+    *******************************************************************************/
+   @BeforeEach
+   void seed() throws Exception
+   {
+      admin();
+      var table = qInstance.getTable("patient");
+      table.withSupplementalMetaData(new AiTableMetaData().withClassification(DataClassification.PHI));
+      table.getField("email").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
+      table.getField("id").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PUBLIC));
+      table.getField("status").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PUBLIC));
+      table.withAuditRules(new QAuditRules().withAuditLevel(AuditLevel.FIELD));
+      config.withClassificationSource(new ConfiguredClassificationSource());
+      assertThat(config.getClassificationSource().ofTable("patient")).isEqualTo(DataClassification.PHI);
+      new UpdateAction().execute(new UpdateInput("patient").withRecord(new QRecord().withValue("id", 1).withValue("email", pii).withValue("diagnosis", phi)));
+      config.getRuntime().getStore().insert(new AssistantProvider().withName("Eligible local provider").withProviderType("patient-test")
+         .withModel("test-model").withLocation("LOCAL").withHipaaEligible(true).withEnabled(true).withIsDefault(true));
+      config.getRuntime().getStore().insert(new AssistantDataAccess().withObjectType("TABLE").withObjectName("patient").withMode("READ_WRITE"));
+      config.getRuntime().getDataAccessService().invalidate();
+   }
+
+
+
+   /*******************************************************************************
+    ** Scan raw backend storage as an administrator, not just decoded transcript APIs.
+    *******************************************************************************/
+   private void admin()
+   {
+      QContext.init(qInstance, AssistantTestFixture.ownerSession(AssistantTestFixture.ADMIN, config));
+   }
+
+
+
+   /*******************************************************************************
+    ** Every five-character window is sufficient to detect any leaked substring > 4.
+    *******************************************************************************/
+   private void assertNoCanaries(String text)
+   {
+      for(String canary : List.of(phi, pii))
+      {
+         for(int i = 0; i <= canary.length() - 5; i++)
+         {
+            assertThat(text).doesNotContain(canary.substring(i, i + 5));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Query every registered assistant table and all core audit tables, without a row cap.
+    *******************************************************************************/
+   private void scanStorage() throws Exception
+   {
+      admin();
+      for(String table : qInstance.getTables().keySet())
+      {
+         if(table.startsWith("assistant_") || table.toLowerCase().startsWith("audit"))
+         {
+            var rows = new QueryAction().execute(new QueryInput(table).withFilter(new QQueryFilter())).getRecords();
+            assertNoCanaries(JsonUtils.toJson(rows));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Query, get, input filters, presentation, split deltas, proposals and errors.
+    ** Log capture includes exception detail and a provider thread without QContext.
+    *******************************************************************************/
+   @Test
+   void canariesNeverReachStorageOrLogs() throws Exception
+   {
+      List<String> logs = new CopyOnWriteArrayList<>();
+      AbstractAppender appender = new AbstractAppender("transcript-canary", null, PatternLayout.createDefaultLayout(), false, Property.EMPTY_ARRAY)
+      {
+         /*******************************************************************************
+          ** Retain the emitted log representation, including any exception.
+          *******************************************************************************/
+         @Override
+         public void append(LogEvent event)
+         {
+            logs.add(event.getMessage().getFormattedMessage() + (event.getThrown() == null ? "" : event.getThrown().toString()));
+         }
+      };
+      appender.start();
+      Logger logger = (Logger) LogManager.getRootLogger();
+      logger.addAppender(appender);
+      try
+      {
+         provider.thenToolUse("query_records", Map.of("table", "patient"))
+            .thenToolUse("get_record", Map.of("table", "patient", "id", "1"))
+            .thenRespond(request ->
+            {
+               assertThat(JsonUtils.toJson(request)).contains(phi, pii);
+               QLogger.getLogger(TranscriptRedactionTest.class).warn("Provider echoed " + phi + " " + pii, new IllegalStateException(phi + " " + pii));
+               return (new ModelResponse().withParts(List.of(ModelPart.toolUse("present", "present_results", Map.of("summary", phi + " " + pii))))
+                  .withStopReason(ModelResponse.STOP_TOOL_USE));
+            })
+            .thenToolUse("query_records", Map.of("table", "patient", "filter", filter("AND", "email", "EQUALS", List.of(pii))))
+            .thenToolUse("update_record", Map.of("table", "patient", "id", "1", "values", Map.of("diagnosis", phi, "email", pii)))
+            .thenText("x".repeat(43) + phi + " " + pii);
+         var outcome = runner.run(AssistantTestFixture.ADMIN, phi + " " + pii);
+         assertThat(outcome.errorCode()).isNull();
+         assertThat(outcome.text()).contains(phi, pii);
+         assertThat(outcome.toolCalls()).hasSize(5).allSatisfy(call -> assertThat(call.getOutcome()).isEqualTo("OK"));
+         admin();
+         var store = config.getRuntime().getStore();
+         var proposals = store.query(AssistantPendingAction.TABLE_NAME, new QQueryFilter(), AssistantPendingAction.class);
+         assertThat(proposals).singleElement().satisfies(action ->
+         {
+            assertThat(action.getParamsJson()).contains("changed:v1");
+            assertThat(action.getStatus()).isEqualTo(AssistantPendingAction.STATUS_REJECTED);
+         });
+         assertThatThrownBy(() -> new AssistantApi(config.getRuntime(), QContext.getQSession()).confirmAction(proposals.getFirst().getId()))
+            .hasMessageContaining("rejected");
+         var events = store.query(AssistantTurnEvent.TABLE_NAME, new QQueryFilter(), AssistantTurnEvent.class);
+         assertThat(events).isNotEmpty();
+         assertThat(events.stream().map(AssistantTurnEvent::getDataJson).reduce("", String::concat)).contains("changed:v1");
+         scanStorage();
+
+         provider.thenToolUse("get_record", Map.of("table", "patient", "id", "1"))
+            .thenError(ModelProviderException.Kind.INVALID_REQUEST, phi + " " + pii);
+         assertThat(runner.run(AssistantTestFixture.ADMIN, "Read patients again").errorCode()).isEqualTo("PROVIDER");
+         scanStorage();
+         assertThat(logs).isNotEmpty();
+         assertNoCanaries(String.join("\n", logs));
+         assertThat(QLogger.redact(phi)).as("turn hook released").isEqualTo(phi);
+      }
+      finally
+      {
+         logger.removeAppender(appender);
+         appender.stop();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Content written before discovery never reaches storage, even transiently.
+    ** Includes the audit reason, scheduled summary, bulk log and redactions payloads.
+    *******************************************************************************/
+   @Test
+   void provisionalPayloadsAndNoResultCriteria() throws Exception
+   {
+      admin();
+      var store = config.getRuntime().getStore();
+      try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
+      {
+         redactor.attach();
+         var audit = store.insert(new AssistantDataAudit().withUserId(AssistantTestFixture.ADMIN).withReason(phi + " " + pii));
+         store.insert(new AssistantScheduledRun().withUserId(AssistantTestFixture.ADMIN).withTaskId(1).withScheduledFor(Instant.now()).withSummary(phi));
+         store.insert(new AssistantToolCall().withUserId(AssistantTestFixture.ADMIN).withToolName("bulk_load_preview")
+            .withInputJson(JsonUtils.toJson(Map.of("email", pii))).withResultJson(phi).withRedactionsJson(JsonUtils.toJson(List.of(phi, pii))));
+         scanStorage();
+         redactor.input(Map.of("table", "patient", "filter", Map.of("subFilters", List.of(filter("AND", "email", "EQUALS", List.of(pii))))));
+         redactor.observe("patient", "diagnosis", 1, phi, phi);
+         store.updateFields(AssistantDataAudit.TABLE_NAME, new QRecord().withValue("id", audit.getId()).withValue("reason", pii + " " + phi));
+         redactor.flush(store);
+         scanStorage();
+      }
+      provider.thenToolUse("query_records", Map.of("table", "patient", "filter", filter("AND", "email", "EQUALS", List.of(pii + "-absent"))))
+         .thenText(pii + "-absent");
+      runner.run(AssistantTestFixture.ADMIN, "Find missing patient");
+      scanStorage();
+   }
+
+
+
+   /*******************************************************************************
+    ** Metadata must not reject or clip tokens after the SQL columns are widened.
+    *******************************************************************************/
+   @Test
+   void expandedTokensSurviveMetadataValidation() throws Exception
+   {
+      admin();
+      var store = config.getRuntime().getStore();
+      String raw = phi.repeat(250);
+      try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
+      {
+         redactor.attach();
+         redactor.observe("patient", "diagnosis", 1, phi, phi);
+         var audit = store.insert(new AssistantDataAudit().withUserId(AssistantTestFixture.ADMIN).withReason(raw));
+         var turn = store.insert(new AssistantTurn().withUserId(AssistantTestFixture.ADMIN).withStatus("RUNNING").withStopReason(raw));
+         var call = store.insert(new AssistantToolCall().withUserId(AssistantTestFixture.ADMIN).withToolCallId(raw).withResultSummary(raw));
+         var usage = store.insert(new AssistantUsage().withUserId(AssistantTestFixture.ADMIN).withRequestId(raw));
+         redactor.flush(store);
+         String expected = "changed:v1".repeat(250);
+         assertThat(store.getTurn(AssistantTestFixture.ADMIN, turn.getId()).getStopReason()).isEqualTo(expected);
+         for(var check : List.of(Map.entry(audit, "reason"), Map.entry(call, "toolCallId"), Map.entry(call, "resultSummary"), Map.entry(usage, "requestId")))
+         {
+            var row = new QueryAction().execute(new QueryInput(check.getKey().tableName())
+               .withFilter(new QQueryFilter().withCriteria("id", com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator.EQUALS,
+                  check.getKey().toQRecord().getValue("id")))).getRecords().getFirst();
+            assertThat(row.getValueString(check.getValue())).isEqualTo(expected);
+         }
+      }
+      scanStorage();
+   }
+
+
+
+   /*******************************************************************************
+    ** A late tool retains the log hook; incomplete/cancelled turns keep placeholders
+    ** and cannot turn a provisional pending payload into an executable empty write.
+    *******************************************************************************/
+   @Test
+   void lateWorkerKeepsProtectionAndFailsClosed() throws Exception
+   {
+      admin();
+      var store = config.getRuntime().getStore();
+      TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource());
+      try
+      {
+         redactor.attach();
+         redactor.observe("patient", "diagnosis", 1, phi, phi);
+         assertThat(redactor.retainWorker()).isTrue();
+         var action = store.insert(new AssistantPendingAction().withUserId(AssistantTestFixture.ADMIN)
+            .withStatus(AssistantPendingAction.STATUS_PENDING).withParamsJson(JsonUtils.toJson(Map.of("diagnosis", phi))));
+         assertThat(store.getPendingAction(AssistantTestFixture.ADMIN, action.getId()).getStatus()).isEqualTo("PREPARING");
+         redactor.flush(store);
+         assertThat(store.getPendingAction(AssistantTestFixture.ADMIN, action.getId()).getStatus()).isEqualTo(AssistantPendingAction.STATUS_REJECTED);
+         redactor.close();
+         assertThat(redactor.retainWorker()).isFalse();
+         assertThat(QLogger.redact(phi)).isEqualTo("changed:v1");
+         redactor.releaseWorker();
+         assertThat(QLogger.redact(phi)).isEqualTo(phi);
+         scanStorage();
+      }
+      finally
+      {
+         redactor.close();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Equal values keep distinct record bindings; canonical decimals and escaped text
+    ** redact exactly and case-sensitively, with no mutation of model-facing objects.
+    *******************************************************************************/
+   @Test
+   void recordBindingsAndCanonicalVariants() throws Exception
+   {
+      admin();
+      qInstance.withSupplementalMetaData(new ClassifiedValueKeySourceMetaData().withKeySourceInstance(new ClassifiedValueKeySource()
+      {
+         /*******************************************************************************
+          ** Deterministic test key, never a production secret.
+          *******************************************************************************/
+         @Override
+         public Optional<KeyAndId> instanceKey()
+         {
+            return (Optional.of(new KeyAndId(new byte[32], "test")));
+         }
+
+
+
+         /*******************************************************************************
+          ** No per-record keys in this instance-key test.
+          *******************************************************************************/
+         @Override
+         public Optional<KeyAndId> keyForRecordField(String table, String field, Serializable pk, boolean create)
+         {
+            return (Optional.empty());
+         }
+      }));
+      var table = qInstance.getTable("patient");
+      try(TranscriptRedactor redactor = new TranscriptRedactor(qInstance, config.getClassificationSource()))
+      {
+         redactor.observe("patient", "diagnosis", 1, phi, phi);
+         redactor.observe("patient", "diagnosis", 2, phi, phi);
+         String first = ClassifiedValueTokens.forRecordField(table, table.getField("diagnosis"), 1, phi);
+         String second = ClassifiedValueTokens.forRecordField(table, table.getField("diagnosis"), 2, phi);
+         assertThat(first).isNotEqualTo(second);
+         String stored = redactor.payload(JsonUtils.toJson(Map.of("table", "patient", "records", List.of(
+            Map.of("id", 1, "values", Map.of("diagnosis", phi)), Map.of("id", 2, "values", Map.of("diagnosis", phi))))));
+         assertThat(stored).contains(first, second).doesNotContain(phi);
+         assertThat(ClassifiedValueTokens.verifyRecordField(table, table.getField("diagnosis"), 1, phi, first)).isTrue();
+         redactor.observe("patient", "diagnosis", 3, new BigDecimal("123.4500"), "123.4500");
+         String numeric = redactor.redact("123.4500 and 123.45");
+         assertThat(numeric).doesNotContain("123.45").contains("hmac:v1:test:");
+         assertThat(redactor.redact(phi.toLowerCase())).isEqualTo(phi.toLowerCase());
+         redactor.variant("patient", "diagnosis", "1", phi.substring(0, 25));
+         assertThat(redactor.redact(phi.substring(0, 25))).isEqualTo(first);
+         redactor.truncated("Detail: " + pii, "Detail: " + pii.substring(0, 20));
+         redactor.observe("patient", "email", 1, pii, pii);
+         redactor.truncated("Detail: " + pii, "Detail: " + pii.substring(0, 20));
+         assertThat(redactor.redact("Detail: " + pii.substring(0, 20))).doesNotContain("CANARY");
+         assertThat(redactor.redact(first)).isEqualTo(first);
+         redactor.observe("patient", "diagnosis", 4, "quote\"line\n", "quote\"line\n");
+         assertThat(redactor.payload(JsonUtils.toJson(Map.of("text", "quote\"line\n")))).doesNotContain("quote");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** No cloud dependency: both rows use the recording provider.
+    *******************************************************************************/
+   class PatientProviderType implements AssistantProviderType
+   {
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public String getType()
+      {
+         return ("patient-test");
+      }
+
+
+
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public String getLabel()
+      {
+         return ("Test");
+      }
+
+
+
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public List<String> getModels()
+      {
+         return (List.of("test-model"));
+      }
+
+
+
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public String getDefaultApiKeyEnvName()
+      {
+         return (null);
+      }
+
+
+
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public String getDefaultBaseUrl()
+      {
+         return ("http://localhost");
+      }
+
+
+
+      /*******************************************************************************
+       ** Fixture adapter.
+       *******************************************************************************/
+      @Override
+      public AssistantModelProvider create(String apiKey, String baseUrl)
+      {
+         return (provider);
+      }
+
+
+
+   }
+}

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/persistence/SchemaForeignKeyPostgresTest.java

@@ -0,0 +1,387 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.assistant.persistence;
+
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.joins.JoinOn;
+import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValueSource;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValueSourceType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.PostgreSQLContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Guardrail for the reference DDL (docs/schema-postgres.sql) on a real
+ ** PostgreSQL: every relation the qbit declares (a table possible-value source
+ ** on an assistant_ field, or a join between assistant_ tables), plus the
+ ** relations only the code knows, has a foreign key with the intended ON DELETE
+ ** rule, and every foreign key column leads an index. A new relation without a
+ ** foreign key fails here unless it is added to EXEMPT with a reason.
+ *******************************************************************************/
+@Testcontainers(disabledWithoutDocker = true)
+class SchemaForeignKeyPostgresTest extends BaseAssistantTest
+{
+   @Container
+   static final PostgreSQLContainer<?> POSTGRES = new PostgreSQLContainer<>(DockerImageName.parse("pgvector/pgvector:pg16").asCompatibleSubstituteFor("postgres"));
+
+   /////////////////////////////////////////////////////////////////////////
+   // relations the meta-data does not declare (no possible-value source) //
+   /////////////////////////////////////////////////////////////////////////
+   private static final List<Relation> CODE_ONLY = List.of(
+      new Relation("assistant_data_audit", "turn_id", "assistant_turn"),
+      new Relation("assistant_data_audit", "provider_id", "assistant_provider"),
+      new Relation("assistant_budget_usage", "provider_id", "assistant_provider"));
+
+   //////////////////////////////////////////////////////////////////
+   // relations that deliberately have no foreign key, with reason //
+   //////////////////////////////////////////////////////////////////
+   private static final Map<Relation, String> EXEMPT = Map.of(
+      new Relation("assistant_budget_usage", "usage_id", "assistant_usage"),
+      "budget accounting projection: it must outlive its usage row so budgets keep counting the spend");
+
+   ///////////////////////////////////////////////////////////////////////////
+   // the intended ON DELETE rule (pg confdeltype: a NO ACTION, c CASCADE,  //
+   // n SET NULL) - see the foreign key block in docs/schema-postgres.sql   //
+   ///////////////////////////////////////////////////////////////////////////
+   private static final Map<Relation, String> RULES = new LinkedHashMap<>();
+
+   static
+   {
+      rule("assistant_message", "conversation_id", "assistant_conversation", "c");
+      rule("assistant_message", "parent_message_id", "assistant_message", "n");
+      rule("assistant_message", "turn_id", "assistant_turn", "n");
+      rule("assistant_turn", "conversation_id", "assistant_conversation", "c");
+      rule("assistant_turn", "user_message_id", "assistant_message", "n");
+      rule("assistant_turn", "assistant_message_id", "assistant_message", "n");
+      rule("assistant_turn_event", "turn_id", "assistant_turn", "c");
+      rule("assistant_tool_call", "conversation_id", "assistant_conversation", "c");
+      rule("assistant_tool_call", "turn_id", "assistant_turn", "c");
+      rule("assistant_tool_call", "pending_action_id", "assistant_pending_action", "n");
+      rule("assistant_feedback", "conversation_id", "assistant_conversation", "c");
+      rule("assistant_feedback", "message_id", "assistant_message", "c");
+      rule("assistant_pending_action", "conversation_id", "assistant_conversation", "c");
+      rule("assistant_pending_action", "turn_id", "assistant_turn", "n");
+      rule("assistant_data_audit", "conversation_id", "assistant_conversation", "n");
+      rule("assistant_data_audit", "turn_id", "assistant_turn", "n");
+      rule("assistant_data_audit", "provider_id", "assistant_provider", "a");
+      rule("assistant_usage", "conversation_id", "assistant_conversation", "n");
+      rule("assistant_usage", "turn_id", "assistant_turn", "n");
+      rule("assistant_usage", "provider_id", "assistant_provider", "a");
+      rule("assistant_budget_usage", "provider_id", "assistant_provider", "a");
+      rule("assistant_profile_skill", "profile_id", "assistant_profile", "c");
+      rule("assistant_profile_skill", "skill_id", "assistant_skill", "c");
+      rule("assistant_model", "provider_id", "assistant_provider", "c");
+      rule("assistant_scheduled_run", "conversation_id", "assistant_conversation", "n");
+      rule("assistant_scheduled_run", "turn_id", "assistant_turn", "n");
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   record Relation(String child, String column, String parent)
+   {
+      @Override
+      public String toString()
+      {
+         return (child + "." + column + " -> " + parent);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void rule(String child, String column, String parent, String rule)
+   {
+      RULES.put(new Relation(child, column, parent), rule);
+   }
+
+
+
+   /*******************************************************************************
+    ** Apply the whole reference DDL, then its foreign key block a second time
+    ** (it must be re-runnable on an existing database).
+    *******************************************************************************/
+   @BeforeAll
+   static void applyDdl() throws Exception
+   {
+      String ddl = Files.readString(Path.of("..", "docs", "schema-postgres.sql"));
+      try(Connection connection = connect(); Statement statement = connection.createStatement())
+      {
+         statement.execute(ddl);
+         int start = ddl.indexOf("-- Foreign keys (2026-10-14)");
+         if(start >= 0)
+         {
+            statement.execute(ddl.substring(start));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Every declared or code-only relation has a foreign key, unless exempt.
+    *******************************************************************************/
+   @Test
+   void everyRelationHasAForeignKey() throws Exception
+   {
+      Set<Relation> relations = new LinkedHashSet<>(declaredRelations());
+      assertThat(relations).as("declared relations found in the meta-data").hasSizeGreaterThan(20);
+      relations.addAll(CODE_ONLY);
+
+      Map<Relation, String> foreignKeys = foreignKeys();
+      List<String>          missing     = new ArrayList<>();
+      for(Relation relation : relations)
+      {
+         if(!foreignKeys.containsKey(relation) && !EXEMPT.containsKey(relation))
+         {
+            missing.add(relation.toString());
+         }
+      }
+      assertThat(missing).as("relations without a foreign key (add one to docs/schema-postgres.sql, or to EXEMPT with a reason)").isEmpty();
+
+      for(Relation exempt : EXEMPT.keySet())
+      {
+         assertThat(foreignKeys).as("exempt relation " + exempt + " must stay without a foreign key").doesNotContainKey(exempt);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The ON DELETE rules match how the qbit deletes.
+    *******************************************************************************/
+   @Test
+   void deleteRulesMatchTheDeletePaths() throws Exception
+   {
+      Map<Relation, String> foreignKeys = foreignKeys();
+      for(Map.Entry<Relation, String> entry : RULES.entrySet())
+      {
+         assertThat(foreignKeys.get(entry.getKey())).as(entry.getKey().toString()).isEqualTo(entry.getValue());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Every foreign key column is the first column of some index.
+    *******************************************************************************/
+   @Test
+   void everyForeignKeyColumnLeadsAnIndex() throws Exception
+   {
+      List<String> unindexed = new ArrayList<>();
+      try(Connection connection = connect(); Statement statement = connection.createStatement();
+         ResultSet rs = statement.executeQuery("""
+            SELECT c.conrelid::regclass::text, a.attname
+            FROM pg_constraint c
+            JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = c.conkey[1]
+            WHERE c.contype = 'f' AND c.connamespace = 'public'::regnamespace
+              AND NOT EXISTS (SELECT 1 FROM pg_index i WHERE i.indrelid = c.conrelid AND i.indkey[0] = c.conkey[1])
+            """))
+      {
+         while(rs.next())
+         {
+            unindexed.add(rs.getString(1) + "." + rs.getString(2));
+         }
+      }
+      assertThat(unindexed).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** AssistantStore.deleteConversation's delete order works under the foreign
+    ** keys; usage (cost accounting) survives with its pointers nulled. A plain
+    ** delete of a conversation (admin screen) cascades through the
+    ** message <-> turn cycle and keeps usage and the data audit.
+    *******************************************************************************/
+   @Test
+   void conversationDeletesWorkUnderTheForeignKeys() throws Exception
+   {
+      try(Connection connection = connect(); Statement statement = connection.createStatement())
+      {
+         connection.setAutoCommit(false);
+         for(int c : new int[] { 1, 2 })
+         {
+            int m = c * 10;
+            statement.execute("INSERT INTO assistant_provider (id, name, provider_type) VALUES (" + c + ", 'p" + c + "', 'anthropic')");
+            statement.execute("INSERT INTO assistant_conversation (id, user_id) VALUES (" + c + ", 'u')");
+            statement.execute("INSERT INTO assistant_message (id, conversation_id, user_id, role) VALUES (" + m + ", " + c + ", 'u', 'USER')");
+            statement.execute("INSERT INTO assistant_turn (id, conversation_id, user_id, user_message_id, provider_id) VALUES (" + c + ", " + c + ", 'u', " + m + ", NULL)");
+            statement.execute("INSERT INTO assistant_message (id, conversation_id, user_id, role, turn_id, parent_message_id) VALUES (" + (m + 1) + ", " + c + ", 'u', 'ASSISTANT', " + c + ", " + m + ")");
+            statement.execute("UPDATE assistant_message SET turn_id = " + c + " WHERE id = " + m);
+            statement.execute("UPDATE assistant_turn SET assistant_message_id = " + (m + 1) + " WHERE id = " + c);
+            statement.execute("INSERT INTO assistant_turn_event (turn_id, user_id) VALUES (" + c + ", 'u')");
+            statement.execute("INSERT INTO assistant_pending_action (id, conversation_id, turn_id, user_id) VALUES (" + c + ", " + c + ", " + c + ", 'u')");
+            statement.execute("INSERT INTO assistant_tool_call (conversation_id, turn_id, pending_action_id, user_id) VALUES (" + c + ", " + c + ", " + c + ", 'u')");
+            statement.execute("INSERT INTO assistant_feedback (conversation_id, message_id, user_id) VALUES (" + c + ", " + (m + 1) + ", 'u')");
+            statement.execute("INSERT INTO assistant_data_audit (id, conversation_id, turn_id, provider_id, user_id) VALUES (" + c + ", " + c + ", " + c + ", " + c + ", 'u')");
+            statement.execute("INSERT INTO assistant_usage (id, conversation_id, turn_id, provider_id, user_id) VALUES (" + c + ", " + c + ", " + c + ", " + c + ", 'u')");
+         }
+
+         ////////////////////////////////////////////////////////////
+         // conversation 1: AssistantStore.deleteConversation order //
+         ////////////////////////////////////////////////////////////
+         statement.execute("DELETE FROM assistant_turn_event WHERE turn_id IN (SELECT id FROM assistant_turn WHERE conversation_id = 1)");
+         for(String table : List.of("assistant_tool_call", "assistant_feedback", "assistant_message", "assistant_turn", "assistant_data_audit", "assistant_pending_action"))
+         {
+            statement.execute("DELETE FROM " + table + " WHERE conversation_id = 1");
+         }
+         statement.execute("DELETE FROM assistant_conversation WHERE id = 1");
+         assertThat(queryString(statement, "SELECT conversation_id || '/' || turn_id FROM assistant_usage WHERE id = 1")).as("usage kept, pointers nulled").isNull();
+         assertThat(queryString(statement, "SELECT provider_id::text FROM assistant_usage WHERE id = 1")).isEqualTo("1");
+
+         //////////////////////////////////////////////
+         // conversation 2: a plain conversation delete //
+         //////////////////////////////////////////////
+         statement.execute("DELETE FROM assistant_conversation WHERE id = 2");
+         for(String table : List.of("assistant_message", "assistant_turn", "assistant_tool_call", "assistant_feedback", "assistant_pending_action"))
+         {
+            assertThat(queryString(statement, "SELECT COUNT(*)::text FROM " + table + " WHERE conversation_id = 2")).as(table).isEqualTo("0");
+         }
+         assertThat(queryString(statement, "SELECT COUNT(*)::text FROM assistant_turn_event")).isEqualTo("0");
+         assertThat(queryString(statement, "SELECT (conversation_id IS NULL AND turn_id IS NULL)::text FROM assistant_data_audit WHERE id = 2")).as("data audit kept").isEqualTo("true");
+         assertThat(queryString(statement, "SELECT (conversation_id IS NULL AND turn_id IS NULL)::text FROM assistant_usage WHERE id = 2")).as("usage kept").isEqualTo("true");
+         connection.rollback();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Possible-value-source fields and joins between assistant_ tables.
+    *******************************************************************************/
+   private List<Relation> declaredRelations()
+   {
+      List<Relation> relations = new ArrayList<>();
+      for(QTableMetaData table : qInstance.getTables().values())
+      {
+         if(!table.getName().startsWith("assistant_"))
+         {
+            continue;
+         }
+         for(QFieldMetaData field : table.getFields().values())
+         {
+            QPossibleValueSource pvs = field.getPossibleValueSourceName() == null ? null : qInstance.getPossibleValueSource(field.getPossibleValueSourceName());
+            if(pvs != null && pvs.getType() == QPossibleValueSourceType.TABLE && pvs.getTableName().startsWith("assistant_"))
+            {
+               relations.add(new Relation(table.getName(), snake(field.getName()), pvs.getTableName()));
+            }
+         }
+      }
+      for(QJoinMetaData join : qInstance.getJoins().values())
+      {
+         if(!join.getLeftTable().startsWith("assistant_") || !join.getRightTable().startsWith("assistant_"))
+         {
+            continue;
+         }
+         for(JoinOn on : join.getJoinOns())
+         {
+            if("id".equals(on.getLeftField()))
+            {
+               relations.add(new Relation(join.getRightTable(), snake(on.getRightField()), join.getLeftTable()));
+            }
+            else if("id".equals(on.getRightField()))
+            {
+               relations.add(new Relation(join.getLeftTable(), snake(on.getLeftField()), join.getRightTable()));
+            }
+         }
+      }
+      return (relations);
+   }
+
+
+
+   /*******************************************************************************
+    ** Single-column foreign keys in the database, with their delete rule.
+    *******************************************************************************/
+   private static Map<Relation, String> foreignKeys() throws Exception
+   {
+      Map<Relation, String> foreignKeys = new LinkedHashMap<>();
+      try(Connection connection = connect(); Statement statement = connection.createStatement();
+         ResultSet rs = statement.executeQuery("""
+            SELECT c.conrelid::regclass::text, a.attname, c.confrelid::regclass::text, c.confdeltype
+            FROM pg_constraint c
+            JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = c.conkey[1]
+            WHERE c.contype = 'f' AND array_length(c.conkey, 1) = 1 AND c.connamespace = 'public'::regnamespace
+            """))
+      {
+         while(rs.next())
+         {
+            foreignKeys.put(new Relation(rs.getString(1), rs.getString(2), rs.getString(3)), rs.getString(4));
+         }
+      }
+      return (foreignKeys);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static String queryString(Statement statement, String sql) throws Exception
+   {
+      try(ResultSet rs = statement.executeQuery(sql))
+      {
+         return (rs.next() ? rs.getString(1) : null);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static String snake(String camel)
+   {
+      return (camel.replaceAll("([A-Z])", "_$1").toLowerCase());
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Connection connect() throws Exception
+   {
+      return (DriverManager.getConnection(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword()));
+   }
+}

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/rag/RagTest.java

@@ -21,6 +21,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import com.kingsrook.qbits.assistant.AssistantQBitConfig;
 import com.kingsrook.qbits.assistant.engine.PolicyService;
 import com.kingsrook.qbits.assistant.engine.TurnClassification;
@@ -38,6 +39,7 @@ import com.kingsrook.qbits.assistant.spi.AssistantProviderType;
 import com.kingsrook.qbits.assistant.spi.EmbeddingRequest;
 import com.kingsrook.qbits.assistant.spi.EmbeddingResponse;
 import com.kingsrook.qbits.assistant.spi.ModelPart;
+import com.kingsrook.qbits.assistant.spi.ModelResponse;
 import com.kingsrook.qbits.assistant.spi.ModelUsage;
 import com.kingsrook.qbits.assistant.testing.AssistantTestFixture;
 import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
@@ -50,12 +52,17 @@ import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
 import com.kingsrook.qqq.backend.core.actions.ai.AiToolOutcome;
 import com.kingsrook.qqq.backend.core.actions.ai.AiToolResult;
 import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.middleware.javalin.specs.invoke.UiApiCredentials;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
@@ -242,6 +249,49 @@ class RagTest extends BaseAssistantTest
 
 
 
+   /*******************************************************************************
+    ** Indexed PHI is re-read live; snippets reach the eligible model but never
+    ** assistant rows or audit values.
+    *******************************************************************************/
+   @Test
+   @Tag("redteam")
+   void classifiedKnowledgeCanary() throws Exception
+   {
+      String phi = "CANARY-PHI-" + UUID.randomUUID();
+      String pii = "CANARY-PII-" + UUID.randomUUID();
+      var table = qInstance.getTable("blog_post");
+      table.getField("content").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PHI));
+      table.getField("title").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
+      new UpdateAction().execute(new UpdateInput("blog_post").withRecord(new QRecord().withValue("id", 1)
+         .withValue("title", pii).withValue("content", phi)));
+      config.getRuntime().getStore().updateFields(AssistantProvider.TABLE_NAME, new QRecord().withValue("id", embeddingProvider.getId())
+         .withValue("hipaaEligible", true).withValue("isDefault", true));
+      assertThat(config.getRuntime().getRagService().index(source).errors()).isZero();
+      embeddings.thenToolUse("search_knowledge", Map.of("query", "Drain upgrade verify", "limit", 8)).thenRespond(request ->
+      {
+         assertThat(JsonUtils.toJson(request)).contains(phi, pii);
+         return (new ModelResponse().withParts(List.of(ModelPart.text(phi + " " + pii))).withStopReason(ModelResponse.STOP_END_TURN));
+      });
+      assertThat(runner.run(AssistantTestFixture.ADMIN, "Search knowledge").errorCode()).isNull();
+      as(AssistantTestFixture.ADMIN);
+      for(String name : qInstance.getTables().keySet())
+      {
+         if(name.startsWith("assistant_") || name.toLowerCase().startsWith("audit"))
+         {
+            String rows = JsonUtils.toJson(new QueryAction().execute(new QueryInput(name)).getRecords());
+            for(String marker : List.of(phi, pii))
+            {
+               for(int i = 0; i <= marker.length() - 5; i++)
+               {
+                  assertThat(rows).doesNotContain(marker.substring(i, i + 5));
+               }
+            }
+         }
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Unchanged content performs no embedding; disappeared records are removed.
     *******************************************************************************/

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/scheduled/ScheduledTasksTest.java

@@ -32,6 +32,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.Executors;
@@ -56,7 +57,9 @@ import com.kingsrook.qbits.assistant.testing.BaseAssistantTest;
 import com.kingsrook.qbits.assistant.testing.UiApiTestServer;
 import com.kingsrook.qbits.assistant.uiapi.UiApiCall;
 import com.kingsrook.qqq.backend.core.actions.ai.DataClassification;
+import com.kingsrook.qqq.backend.core.actions.ai.AiFieldMetaData;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
@@ -64,12 +67,14 @@ import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutp
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.security.RecordSecurityLock;
 import com.kingsrook.qqq.backend.core.model.session.QSession;
 import com.kingsrook.qqq.backend.core.modules.authentication.implementations.MockAuthenticationModule;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -213,6 +218,35 @@ class ScheduledTasksTest extends BaseAssistantTest
 
 
 
+   /*******************************************************************************
+    ** Scheduled summaries are derived from the redacted stored message, not live SSE.
+    *******************************************************************************/
+   @Test
+   void scheduledSummaryCanary() throws Exception
+   {
+      String marker = "CANARY-PII-" + UUID.randomUUID();
+      qInstance.getTable("page").getField("title").withSupplementalMetaData(new AiFieldMetaData().withClassification(DataClassification.PII));
+      new UpdateAction().execute(new UpdateInput("page").withRecord(new QRecord().withValue("id", 1).withValue("title", marker)));
+      provider.thenToolUse("get_record", Map.of("table", "page", "id", "1")).thenText(marker);
+      AssistantScheduledRun run = scheduledRunner().run(task(), false);
+      assertThat(run.getStatus()).as(run.getError()).isEqualTo("OK");
+      assertThat(run.getSummary()).isEqualTo("changed:v1");
+      assertThat(JsonUtils.toJson(provider.getRequests().getLast())).contains(marker);
+      for(String table : qInstance.getTables().keySet())
+      {
+         if(table.startsWith("assistant_") || table.toLowerCase().startsWith("audit"))
+         {
+            String rows = JsonUtils.toJson(new QueryAction().execute(new QueryInput(table)).getRecords());
+            for(int i = 0; i <= marker.length() - 5; i++)
+            {
+               assertThat(rows).doesNotContain(marker.substring(i, i + 5));
+            }
+         }
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Every tool HTTP call carries both the designated UUID and client channel.
     *******************************************************************************/

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/testing/AssistantTestFixture.java

@@ -99,6 +99,7 @@ public class AssistantTestFixture
     *******************************************************************************/
    public static QInstance build(AssistantQBitConfig config) throws Exception
    {
+      config.setRegisterShutdownHook(false);
       MemoryRecordStore.fullReset();
       QContext.clear();
 

qbit-assistant/src/test/java/com/kingsrook/qbits/assistant/testing/TurnRunner.java

@@ -173,7 +173,7 @@ public class TurnRunner
          {
             last = batch.get(batch.size() - 1).getSeq();
          }
-         if(batch.isEmpty() && running.getLog().isClosed())
+         if(batch.isEmpty() && running.getLog().isClosed() && running.getLog().size() <= last)
          {
             break;
          }