fix: restore safe compaction queueing to main (#58)

docs/pi.md

@@ -38,6 +38,8 @@ This matrix describes Tau's extension compatibility with Pi 0.84.1. Queue contro
 
 While a saved session is working, Tau sends ordinary `prompt` requests with `streamingBehavior: "steer"` or `"followUp"`. Pi decides whether the message is queued or starts an ordinary run if it became idle. Steering messages are delivered as separate user messages together at Pi's next eligible turn boundary; follow-ups run one at a time after the current work. Shift+Enter inserts a newline. Idle submissions use the ordinary prompt path; extension slash commands are not run immediately from the busy queue path.
 
+Queueing remains available during threshold or overflow compaction only when Tau observed compaction begin during an active run. Manual, pre-run automatic, and unknown compaction (including compaction discovered only from a state snapshot) keep the composer editable but block sending. The draft can be sent deliberately after compaction and transcript reconciliation finish; Tau never replays it automatically. Queue admission is checked again after asynchronous mode preparation. If Pi's state changed, Tau recovers the draft instead of falling back to an ordinary prompt. Once a request crosses the bridge, Pi may become idle before handling it; Tau waits for its acknowledgement without resending or projecting an ordinary transcript row.
+
 Before a session's first queue submission, Tau checks Pi's `get_state` modes and sets steering to `all` and follow-ups to `one-at-a-time` if necessary. Pi persists these mode changes as defaults for the relevant local or remote account; they can affect other Pi processes. Tau does not restore old modes because doing so could race with those processes.
 
 **Clear All** issues Pi's `clear_queue` once for both groups. It does not stop the current run, undo messages Pi already selected for delivery, or re-add remaining text. A rejected command shows an in-place upgrade/retry message rather than simulating removal. **Stop** aborts only current work and never clears either queue. Pending messages survive switching between live, warm sessions, but are not stored by Tau or automatically replayed after a process exit, replacement, reconnect, or app restart. Lost queue work gets session-local feedback. A queued but idle runtime is protected from idle eviction.

src/components/ComposerBar.vue

@@ -56,6 +56,14 @@
         :max-width="110"
         @update:model-value="handleEffortChange"
       />
+      <span
+        v-if="compactionRestriction"
+        class="compaction-restriction"
+        role="status"
+        :title="compactionRestriction"
+      >
+        {{ compactionRestriction }}
+      </span>
       <UiTooltip
         v-if="retryPresentation"
         :text="retryPresentation.reason"
@@ -172,6 +180,7 @@ const {
   canCompose,
   canDraft,
   canQueue,
+  compactionRestriction,
   commands,
   compacting,
   currentEffort,
@@ -464,6 +473,16 @@ defineExpose({
   color: var(--faint);
 }
 
+.compaction-restriction {
+  flex: 1;
+  min-width: 0;
+  overflow: hidden;
+  color: var(--faint);
+  font-size: var(--text-xs);
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
 .composer-toolbar {
   display: flex;
   align-items: center;

src/composables/state.test.ts

@@ -688,6 +688,179 @@ describe('fallback feedback', () => {
   });
 });
 
+describe('compaction queue admission', () => {
+  it.each(['threshold', 'overflow'] as const)(
+    'allows active automatic %s compaction but not pre-run compaction',
+    async (compactionReason) => {
+      const { controllerCanQueue } = await import('./state');
+      const controller = testController({
+        ready: true,
+        streaming: true,
+        compacting: true,
+        compactionReason,
+        compactionStartedWhileStreaming: true,
+      });
+      expect(controllerCanQueue(controller)).toBe(true);
+      controller.compactionStartedWhileStreaming = false;
+      expect(controllerCanQueue(controller)).toBe(false);
+      controller.compacting = false;
+      expect(controllerCanQueue(controller)).toBe(true);
+    },
+  );
+
+  it.each(['manual', undefined] as const)(
+    'denies compaction reason %s even during an active stream',
+    async (compactionReason) => {
+      const { controllerCanQueue } = await import('./state');
+      expect(
+        controllerCanQueue(
+          testController({
+            ready: true,
+            streaming: true,
+            compacting: true,
+            compactionReason,
+            compactionStartedWhileStreaming: true,
+          }),
+        ),
+      ).toBe(false);
+    },
+  );
+
+  it.each([
+    { ready: false },
+    { disposed: true },
+    { phantom: true },
+    { streaming: false },
+    { stopping: true },
+    { compactionReconciliationPending: true },
+    { starting: true },
+    { promptSubmitting: true },
+    { queueClearing: true },
+    { remoteDisconnected: true },
+  ] satisfies Partial<SessionController>[])(
+    'automatic compaction does not bypass the admission guard %j',
+    async (flags) => {
+      const { controllerCanQueue } = await import('./state');
+      expect(
+        controllerCanQueue(
+          testController({
+            ready: true,
+            streaming: true,
+            compacting: true,
+            compactionReason: 'threshold',
+            compactionStartedWhileStreaming: true,
+            ...flags,
+          }),
+        ),
+      ).toBe(false);
+    },
+  );
+
+  it('keeps ordinary compose blocked through queue setup, unacknowledged submission, and compaction', async () => {
+    const {
+      state,
+      canCompose,
+      canDraft,
+      canQueue,
+      compactionRestriction,
+      queueBusy,
+    } = await import('./state');
+    const controller = testController({ ready: true });
+    const session: SessionSummary = {
+      id: controller.sessionId,
+      path: controller.sessionPath,
+      title: 'Session',
+      lastActive: 'now',
+      lastUserMessageAt: 0,
+      sortAt: 1,
+      archived: false,
+      selected: true,
+    };
+    state.workspace = {
+      activeProjectPath: controller.projectPath,
+      piPath: '/usr/local/bin/pi',
+      projects: [
+        {
+          path: controller.projectPath,
+          name: 'Project',
+          workingDirectory: controller.projectPath,
+          collapsed: false,
+          selected: true,
+          sessions: [session],
+        },
+      ],
+    };
+    state.controllers = [controller];
+    state.activeProjectPath = controller.projectPath;
+    state.activeSessionId = controller.sessionId;
+    state.activeControllerKey = controller.key;
+    const stored = state.controllers[0]!;
+    expect(canCompose.value).toBe(true);
+    expect(canDraft.value).toBe(true);
+    expect(canQueue.value).toBe(false);
+
+    stored.queuePreparing = true;
+    expect(queueBusy.value).toBe(true);
+    expect(canCompose.value).toBe(false);
+    stored.queueSubmissions.push({
+      id: 'queue-prompt',
+      text: 'same',
+      draft: ' same ',
+      kind: 'steer',
+      version: 0,
+      baselineCount: 0,
+    });
+    stored.queuePreparing = false;
+    expect(queueBusy.value).toBe(true);
+    expect(canCompose.value).toBe(false);
+    stored.queueSubmissions = [];
+    expect(queueBusy.value).toBe(false);
+    expect(canCompose.value).toBe(true);
+
+    for (const reason of ['threshold', 'overflow'] as const) {
+      Object.assign(stored, {
+        compacting: true,
+        streaming: true,
+        compactionReason: reason,
+        compactionStartedWhileStreaming: true,
+      });
+      expect(canQueue.value).toBe(true);
+      expect(canCompose.value).toBe(false);
+      expect(canDraft.value).toBe(true);
+      expect(compactionRestriction.value).toBe('');
+      for (const patch of [
+        { compactionReason: 'manual' },
+        { compactionReason: undefined },
+        { compactionStartedWhileStreaming: false },
+        { stopping: true },
+        { compactionReconciliationPending: true },
+      ] satisfies Partial<SessionController>[]) {
+        Object.assign(stored, patch);
+        expect(canQueue.value).toBe(false);
+        expect(canCompose.value).toBe(false);
+        expect(canDraft.value).toBe(true);
+        expect(compactionRestriction.value).toContain('Sending is unavailable');
+        Object.assign(stored, {
+          compactionReason: reason,
+          compactionStartedWhileStreaming: true,
+          stopping: false,
+          compactionReconciliationPending: false,
+        });
+      }
+    }
+    Object.assign(stored, {
+      streaming: false,
+      compacting: false,
+      compactionReconciliationPending: true,
+    });
+    expect(compactionRestriction.value).toBe('');
+    expect(canQueue.value).toBe(false);
+    expect(canCompose.value).toBe(false);
+    stored.compactionReconciliationPending = false;
+    expect(canCompose.value).toBe(true);
+  });
+});
+
 describe('queue recovery identity', () => {
   it('holds an interrupted predecessor draft until that session is reopened', async () => {
     const { ensureController, stashInterruptedQueueDrafts, state } =

src/composables/state.ts

@@ -191,6 +191,8 @@ interface SessionController {
   ready: boolean;
   streaming: boolean;
   compacting: boolean;
+  compactionReason: 'manual' | 'threshold' | 'overflow' | undefined;
+  compactionStartedWhileStreaming: boolean;
 
   compactionReconciliationPending: boolean;
 
@@ -579,23 +581,38 @@ const queueBusy = computed(() =>
 const queueFailedDrafts = computed(
   () => activeController.value?.queueFailedDrafts ?? [],
 );
-const canQueue = computed(() => {
-  const controller = activeController.value;
+function controllerCanQueue(controller: SessionController): boolean {
   return Boolean(
-    controller &&
-    canDraft.value &&
     controller.ready &&
+    !controller.disposed &&
     !controller.phantom &&
     controller.streaming &&
     !controller.stopping &&
-    !controller.compacting &&
+    (!controller.compacting ||
+      (controller.compactionStartedWhileStreaming &&
+        (controller.compactionReason === 'threshold' ||
+          controller.compactionReason === 'overflow'))) &&
     !controller.compactionReconciliationPending &&
     !controller.starting &&
+    !controller.promptSubmitting &&
     !controller.pendingPrompt &&
     !controller.submittedPrompt &&
     !controller.queueClearing &&
     !controller.remoteDisconnected,
   );
+}
+const canQueue = computed(() =>
+  Boolean(
+    activeController.value &&
+    canDraft.value &&
+    controllerCanQueue(activeController.value),
+  ),
+);
+const compactionRestriction = computed(() => {
+  const controller = activeController.value;
+  return controller?.compacting && !controllerCanQueue(controller)
+    ? 'Sending is unavailable during this compaction. Send your draft after it finishes.'
+    : '';
 });
 const compacting = computed(() => activeController.value?.compacting === true);
 const stopping = computed(() => activeController.value?.stopping === true);
@@ -670,6 +687,12 @@ const canCompose = computed(() => {
   if (
     controller.starting ||
     controller.working ||
+    controller.streaming ||
+    controller.stopping ||
+    controller.compacting ||
+    controller.compactionReconciliationPending ||
+    controller.queuePreparing ||
+    controller.queueSubmissions.length > 0 ||
     controller.promptSubmitting ||
     admittingOrdinaryPrompt ||
     projectActionsDisabled.value
@@ -1067,6 +1090,8 @@ function createController(
     ready: false,
     streaming: false,
     compacting: false,
+    compactionReason: undefined,
+    compactionStartedWhileStreaming: false,
     compactionReconciliationPending: false,
     compactionStreamSequence: 0,
     messagesHydrationSequence: 0,
@@ -1694,6 +1719,8 @@ export {
   queueBusy,
   queueFailedDrafts,
   canQueue,
+  controllerCanQueue,
+  compactionRestriction,
   compacting,
   stopping,
   promptSubmitting,

src/composables/useTau/index.ts

@@ -12,6 +12,7 @@ import {
   canCompose,
   canDraft,
   canQueue,
+  compactionRestriction,
   activeQueue,
   queueBusy,
   queueFeedback,
@@ -147,6 +148,7 @@ function useTau() {
     canDraft,
     canCompose,
     canQueue,
+    compactionRestriction,
     activeQueue,
     queueBusy,
     queueFeedback,