chore: add worker stream reconcile dry run

.runtime/portal/RUNBOOK.txt

@@ -36,6 +36,8 @@ Deployment and operations transport:
 Streaming runtime operations:
   node scripts/check-stream-runtime.mjs
   node scripts/runtime-worker-drain.mjs status
+  node scripts/runtime-worker-drain.mjs reconcile
+  node scripts/runtime-worker-drain.mjs reconcile --apply
   node scripts/runtime-worker-drain.mjs drain goosed-3
   node scripts/runtime-worker-drain.mjs undrain goosed-3
   node scripts/check-tool-runtime.mjs

.runtime/portal/scripts/runtime-worker-drain.mjs

@@ -16,7 +16,7 @@ function loadEnvFile(filePath) {
   }
 }
 
-loadEnvFile(path.join(process.cwd(), '.env'));
+loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env'));
 
 const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0';
 const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime';
@@ -26,17 +26,24 @@ const configuredWorkers = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_
   .filter(Boolean)
   .map((_, index) => `goosed-${index + 1}`);
 const action = process.argv[2] || 'status';
-const workerId = process.argv[3] || null;
+const positionalArgs = process.argv.slice(3).filter((arg) => !arg.startsWith('--'));
+const workerId = positionalArgs[0] || null;
+const apply = process.argv.includes('--apply');
+const staleMsArg = process.argv.find((arg) => arg.startsWith('--stale-ms='));
+const staleMs = Math.max(
+  60_000,
+  Number(staleMsArg?.split('=').at(1) || process.env.MEMIND_RUNTIME_STALE_STREAM_MS || 15 * 60 * 1000),
+);
 
 function usage() {
-  console.error('Usage: node scripts/runtime-worker-drain.mjs <status|drain|undrain> [goosed-N]');
+  console.error('Usage: node scripts/runtime-worker-drain.mjs <status|drain|undrain|reconcile> [goosed-N] [--apply] [--stale-ms=900000]');
 }
 
 function workerKey(id, field) {
   return [namespace, 'worker', id, field].join(':');
 }
 
-if (!['status', 'drain', 'undrain'].includes(action)) {
+if (!['status', 'drain', 'undrain', 'reconcile'].includes(action)) {
   usage();
   process.exit(2);
 }
@@ -68,6 +75,7 @@ const workers = [
 if (workerId && !workers.includes(workerId)) workers.push(workerId);
 
 const rows = [];
+const now = Date.now();
 for (const id of [...new Set(workers)].sort()) {
   const values = await client.mGet([
     workerKey(id, 'active_streams'),
@@ -77,16 +85,41 @@ for (const id of [...new Set(workers)].sort()) {
     workerKey(id, 'stream_error_count'),
     workerKey(id, 'last_stream_started_at'),
     workerKey(id, 'last_stream_ended_at'),
+    workerKey(id, 'last_stream_reconciled_at'),
+    workerKey(id, 'stream_reconcile_count'),
   ]);
+  const activeStreams = Number(values[0] || 0);
+  const lastStreamStartedAt = values[5] ? Number(values[5]) : null;
+  const lastStreamEndedAt = values[6] ? Number(values[6]) : null;
+  const staleAgeMs = lastStreamStartedAt ? now - lastStreamStartedAt : null;
+  const stale = Boolean(
+    activeStreams > 0 &&
+    lastStreamStartedAt &&
+    staleAgeMs > staleMs &&
+    (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt),
+  );
+  if (action === 'reconcile' && stale && apply) {
+    await client
+      .multi()
+      .set(workerKey(id, 'active_streams'), '0')
+      .set(workerKey(id, 'last_stream_reconciled_at'), String(now))
+      .incr(workerKey(id, 'stream_reconcile_count'))
+      .exec();
+  }
   rows.push({
     id,
-    activeStreams: Number(values[0] || 0),
+    activeStreams,
     drain: /^(1|true|yes)$/i.test(String(values[1] || '')),
     streamOpenCount: Number(values[2] || 0),
     streamAbortCount: Number(values[3] || 0),
     streamErrorCount: Number(values[4] || 0),
-    lastStreamStartedAt: values[5] ? Number(values[5]) : null,
+    lastStreamStartedAt,
-    lastStreamEndedAt: values[6] ? Number(values[6]) : null,
+    lastStreamEndedAt,
+    lastStreamReconciledAt: values[7] ? Number(values[7]) : null,
+    streamReconcileCount: Number(values[8] || 0),
+    stale,
+    staleAgeMs,
+    reconciled: action === 'reconcile' && stale && apply,
   });
 }
 
@@ -97,5 +130,7 @@ console.log(JSON.stringify({
   action,
   workerId,
   namespace,
+  dryRun: action === 'reconcile' ? !apply : undefined,
+  staleMs: action === 'reconcile' ? staleMs : undefined,
   workers: rows,
 }, null, 2));

scripts/build-portal-runtime.mjs

@@ -324,6 +324,8 @@ async function writeMetadata() {
       'Streaming runtime operations:',
       '  node scripts/check-stream-runtime.mjs',
       '  node scripts/runtime-worker-drain.mjs status',
+      '  node scripts/runtime-worker-drain.mjs reconcile',
+      '  node scripts/runtime-worker-drain.mjs reconcile --apply',
       '  node scripts/runtime-worker-drain.mjs drain goosed-3',
       '  node scripts/runtime-worker-drain.mjs undrain goosed-3',
       '  node scripts/check-tool-runtime.mjs',

scripts/runtime-worker-drain.mjs

@@ -16,7 +16,7 @@ function loadEnvFile(filePath) {
   }
 }
 
-loadEnvFile(path.join(process.cwd(), '.env'));
+loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env'));
 
 const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0';
 const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime';
@@ -26,17 +26,24 @@ const configuredWorkers = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_
   .filter(Boolean)
   .map((_, index) => `goosed-${index + 1}`);
 const action = process.argv[2] || 'status';
-const workerId = process.argv[3] || null;
+const positionalArgs = process.argv.slice(3).filter((arg) => !arg.startsWith('--'));
+const workerId = positionalArgs[0] || null;
+const apply = process.argv.includes('--apply');
+const staleMsArg = process.argv.find((arg) => arg.startsWith('--stale-ms='));
+const staleMs = Math.max(
+  60_000,
+  Number(staleMsArg?.split('=').at(1) || process.env.MEMIND_RUNTIME_STALE_STREAM_MS || 15 * 60 * 1000),
+);
 
 function usage() {
-  console.error('Usage: node scripts/runtime-worker-drain.mjs <status|drain|undrain> [goosed-N]');
+  console.error('Usage: node scripts/runtime-worker-drain.mjs <status|drain|undrain|reconcile> [goosed-N] [--apply] [--stale-ms=900000]');
 }
 
 function workerKey(id, field) {
   return [namespace, 'worker', id, field].join(':');
 }
 
-if (!['status', 'drain', 'undrain'].includes(action)) {
+if (!['status', 'drain', 'undrain', 'reconcile'].includes(action)) {
   usage();
   process.exit(2);
 }
@@ -68,6 +75,7 @@ const workers = [
 if (workerId && !workers.includes(workerId)) workers.push(workerId);
 
 const rows = [];
+const now = Date.now();
 for (const id of [...new Set(workers)].sort()) {
   const values = await client.mGet([
     workerKey(id, 'active_streams'),
@@ -77,16 +85,41 @@ for (const id of [...new Set(workers)].sort()) {
     workerKey(id, 'stream_error_count'),
     workerKey(id, 'last_stream_started_at'),
     workerKey(id, 'last_stream_ended_at'),
+    workerKey(id, 'last_stream_reconciled_at'),
+    workerKey(id, 'stream_reconcile_count'),
   ]);
+  const activeStreams = Number(values[0] || 0);
+  const lastStreamStartedAt = values[5] ? Number(values[5]) : null;
+  const lastStreamEndedAt = values[6] ? Number(values[6]) : null;
+  const staleAgeMs = lastStreamStartedAt ? now - lastStreamStartedAt : null;
+  const stale = Boolean(
+    activeStreams > 0 &&
+    lastStreamStartedAt &&
+    staleAgeMs > staleMs &&
+    (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt),
+  );
+  if (action === 'reconcile' && stale && apply) {
+    await client
+      .multi()
+      .set(workerKey(id, 'active_streams'), '0')
+      .set(workerKey(id, 'last_stream_reconciled_at'), String(now))
+      .incr(workerKey(id, 'stream_reconcile_count'))
+      .exec();
+  }
   rows.push({
     id,
-    activeStreams: Number(values[0] || 0),
+    activeStreams,
     drain: /^(1|true|yes)$/i.test(String(values[1] || '')),
     streamOpenCount: Number(values[2] || 0),
     streamAbortCount: Number(values[3] || 0),
     streamErrorCount: Number(values[4] || 0),
-    lastStreamStartedAt: values[5] ? Number(values[5]) : null,
+    lastStreamStartedAt,
-    lastStreamEndedAt: values[6] ? Number(values[6]) : null,
+    lastStreamEndedAt,
+    lastStreamReconciledAt: values[7] ? Number(values[7]) : null,
+    streamReconcileCount: Number(values[8] || 0),
+    stale,
+    staleAgeMs,
+    reconciled: action === 'reconcile' && stale && apply,
   });
 }
 
@@ -97,5 +130,7 @@ console.log(JSON.stringify({
   action,
   workerId,
   namespace,
+  dryRun: action === 'reconcile' ? !apply : undefined,
+  staleMs: action === 'reconcile' ? staleMs : undefined,
   workers: rows,
 }, null, 2));