feat(shell): redis console store with cross-language streaming integ (#795)

* feat(shell): redis console store with cross-language streaming integ

* feat(shell): console config block, shared console battery, redis store package

* refactor(shell): job_table package, and codex fixes on the redis console

Split job_table into types/constants/table in both languages, mirroring
the console package.

Redis console review fixes: the minted key prefix is public (JobConsole
.store plus RedisConsoleStore.key_prefix) so an external reader can be
handed a console's address; keys expire ttl_seconds after the last
append (default one day) instead of accumulating; the ending chunk is
terminal in the store itself, so an emit racing a kill past the local
guard is dropped server-side; and the TS loader validates the console
block's value types the way Pydantic already did.

Author: zechengz

integ/console/jobs.json

@@ -0,0 +1,60 @@
+{
+  "cases": [
+    {
+      "id": "console_bg_wait_adopts_output",
+      "seq": 509000,
+      "targets": [
+        "ram",
+        "console-redis"
+      ],
+      "command": "(echo out; echo err 1>&2) & wait",
+      "expect": {
+        "exit": 0,
+        "stdout": "out\n",
+        "stderr": "err\n"
+      }
+    },
+    {
+      "id": "console_bg_wait_exit_code",
+      "seq": 509002,
+      "targets": [
+        "ram",
+        "console-redis"
+      ],
+      "command": "(sh -c 'exit 4' & wait $!; echo rc=$?) 2>/dev/null",
+      "expect": {
+        "exit": 0,
+        "stdout": "rc=4\n",
+        "stderr": ""
+      }
+    },
+    {
+      "id": "console_kill_by_bg_id",
+      "seq": 509004,
+      "targets": [
+        "ram",
+        "console-redis"
+      ],
+      "command": "(sleep 5 & kill $! && echo killed) 2>/dev/null",
+      "expect": {
+        "exit": 0,
+        "stdout": "killed\n",
+        "stderr": ""
+      }
+    },
+    {
+      "id": "console_bg_write_lands",
+      "seq": 509006,
+      "targets": [
+        "ram",
+        "console-redis"
+      ],
+      "command": "(echo payload > /data/console-bg.txt & wait) 2>/dev/null; cat /data/console-bg.txt; rm /data/console-bg.txt",
+      "expect": {
+        "exit": 0,
+        "stdout": "payload\n",
+        "stderr": ""
+      }
+    }
+  ]
+}

integ/fixtures/config/accepted.json

@@ -90,6 +90,27 @@
         "mounts": { "/d": { "resource": "ram" } },
         "clis": { "tool": { "script": "tool.py", "runtime": "py", "config": { "token": "t" } } }
       }
+    },
+    {
+      "name": "console block: ram form",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "ram" }
+      }
+    },
+    {
+      "name": "console block: redis form, every key",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "redis", "url": "redis://localhost:6379/5", "key_prefix": "c:", "ttl_seconds": 3600 }
+      }
+    },
+    {
+      "name": "console block: null ttl keeps keys forever",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "redis", "ttl_seconds": null }
+      }
     }
   ]
 }

integ/fixtures/config/rejected.json

@@ -101,6 +101,34 @@
         "mounts": { "/d": { "resource": "ram" } },
         "store": { "type": "ram", "observer": { "type": "s3", "bucket": "b" } }
       }
+    },
+    {
+      "name": "console block: unknown key",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "redis", "prefix": "c:" }
+      }
+    },
+    {
+      "name": "console block: unknown type",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "disk" }
+      }
+    },
+    {
+      "name": "console block: url not a string",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "redis", "url": 123 }
+      }
+    },
+    {
+      "name": "console block: zero ttl",
+      "config": {
+        "mounts": { "/d": { "resource": "ram" } },
+        "console": { "type": "redis", "ttl_seconds": 0 }
+      }
     }
   ]
 }

integ/runners/python/adapters.py

@@ -110,6 +110,9 @@
 from mirage.resource.tencent import TencentConfig, TencentResource
 from mirage.resource.trello import TrelloConfig, TrelloResource
 from mirage.resource.wasabi import WasabiConfig, WasabiResource
+from mirage.shell.console import JobConsole
+from mirage.shell.console.redis import RedisConsoleStore
+from mirage.shell.job_table import ConsoleFactory
 from mirage.types import ConsistencyPolicy
 
 REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
@@ -2273,6 +2276,42 @@ async def teardown_target(
         await service.teardown()
 
 
+def _redis_console(url: str, prefix: str, job_id: int) -> JobConsole:
+    """One job's console on its own Redis stream.
+
+    The nonce beside the id matters because battery cases reap jobs and
+    ids restart at 1; a reused stream would replay the previous case's
+    chunks, ending chunk included.
+
+    Args:
+        url (str): Redis connection URL.
+        prefix (str): this run's key namespace.
+        job_id (int): the job the console is being built for.
+    """
+    key_prefix = f"{prefix}{uuid.uuid4().hex[:8]}-{job_id}:"
+    # Battery keys must not accumulate in the shared redis db.
+    return JobConsole(store=RedisConsoleStore(
+        url=url, key_prefix=key_prefix, ttl_seconds=3600))
+
+
+def console_factory(target: dict, run_id: str) -> ConsoleFactory | None:
+    """Build the target's console factory, or None for in-memory.
+
+    A target opts in with ``"console": {"type": "redis"}``; the stream
+    keys ride REDIS_URL under a per-run namespace.
+
+    Args:
+        target (dict): the target manifest entry.
+        run_id (str): this open's unique id.
+    """
+    block = target.get("console")
+    if block is None:
+        return None
+    url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
+    return functools.partial(_redis_console, url,
+                             f"mirage-integ-console-{run_id}:")
+
+
 async def open_target(
     target: dict,
     consistency: ConsistencyPolicy | None = None
@@ -2281,13 +2320,18 @@ async def open_target(
     service = await make_service(target, run_id)
     mounts, cleanups = await build_mounts(target, run_id, service)
     agent_id = target.get("agentId")
+    factory = console_factory(target, run_id)
     if consistency is not None:
         ws = Workspace(mounts,
                        mode=MountMode.WRITE,
                        consistency=consistency,
-                       agent_id=agent_id)
+                       agent_id=agent_id,
+                       console_factory=factory)
     else:
-        ws = Workspace(mounts, mode=MountMode.WRITE, agent_id=agent_id)
+        ws = Workspace(mounts,
+                       mode=MountMode.WRITE,
+                       agent_id=agent_id,
+                       console_factory=factory)
     for cli_name in target.get("clis", []):
         spec, config = cli_install(service, cli_name)
         ws.register_cli(cli_name, spec, config)

integ/runners/python/harness.py

@@ -24,7 +24,8 @@
 
 # integ/runtime holds the runtime suite (its own schema and runners,
 # integ/runtime/run.{py,ts} + cli.sh), not battery cases; keep it out.
-CASE_DIRS = ("unix", "bash", "crossmount", "resources", "cli", "session")
+CASE_DIRS = ("unix", "bash", "crossmount", "resources", "cli", "session",
+             "console")
 
 
 def integ_root() -> Path:

integ/runners/typescript/adapters.ts

@@ -12,6 +12,7 @@
 // limitations under the License.
 // ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
 
+import { randomBytes } from 'node:crypto'
 import { mkdtempSync, readFileSync, rmSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join, relative, sep } from 'node:path'
@@ -57,6 +58,7 @@ import {
   SLACK,
   HfBucketsResource,
   JaegerResource,
+  JobConsole,
   LanceDBResource,
   LangfuseResource,
   LinearResource,
@@ -74,6 +76,7 @@ import {
   QingStorResource,
   R2Resource,
   RAMResource,
+  RedisConsoleStore,
   RedisResource,
   type Resource,
   S3Resource,
@@ -87,6 +90,7 @@ import {
   TrelloResource,
   WasabiResource,
   Workspace,
+  type ConsoleFactory,
 } from '@struktoai/mirage-node'
 import * as lancedb from '@lancedb/lancedb'
 import { QdrantClient } from '@qdrant/js-client-rest'
@@ -179,6 +183,28 @@ function installLocalClis(ws: { registerCli: (name: string, spec: unknown) => vo
   if (target.clis?.includes('git') === true) ws.registerCli('git', GIT)
 }
 
+// Where a target declares console: {type: 'redis'}, each job's console
+// rides its own Redis stream on REDIS_URL. The nonce beside the id
+// matters because battery cases reap jobs and ids restart at 1; a
+// reused stream would replay the previous case's chunks, ending chunk
+// included. Only the ram opener consults this (main.ts refuses a
+// console block on any other resource), and ws.close() releases the
+// clients through JobTable.closeConsoles.
+function consoleFactoryFor(target: Target): ConsoleFactory | undefined {
+  if (target.console?.type !== 'redis') return undefined
+  const url = process.env.REDIS_URL ?? 'redis://localhost:6379/0'
+  const prefix = `mirage-integ-console-${randomBytes(4).toString('hex')}:`
+  return (jobId: number) =>
+    new JobConsole(
+      new RedisConsoleStore({
+        url,
+        keyPrefix: `${prefix}${randomBytes(4).toString('hex')}-${jobId.toString()}:`,
+        // Battery keys must not accumulate in the shared redis db.
+        ttlSeconds: 3600,
+      }),
+    )
+}
+
 async function openRam(target: Target): Promise<Open> {
   const mounts: Record<string, RAMResource | [RAMResource, MountMode]> = {}
   const built: Record<string, RAMResource> = {}
@@ -194,9 +220,11 @@ async function openRam(target: Target): Promise<Open> {
     built[m.path] = resource
     mounts[m.path] = m.mode === 'read' ? [resource, MountMode.READ] : resource
   }
+  const consoleFactory = consoleFactoryFor(target)
   const ws = new Workspace(mounts, {
     mode: MountMode.WRITE,
     ...(target.agentId !== undefined ? { agentId: target.agentId } : {}),
+    ...(consoleFactory !== undefined ? { consoleFactory } : {}),
   })
   installLocalClis(ws, target)
   return { ws: ws as unknown as ExecWorkspace, cleanup: () => ws.close() }

integ/runners/typescript/harness.ts

@@ -20,7 +20,7 @@ import { fileURLToPath } from 'node:url'
 
 // integ/runtime holds the runtime suite (its own schema and runners,
 // integ/runtime/run.{py,ts} + cli.sh), not battery cases; keep it out.
-const CASE_DIRS = ['unix', 'bash', 'crossmount', 'resources', 'cli', 'session']
+const CASE_DIRS = ['unix', 'bash', 'crossmount', 'resources', 'cli', 'session', 'console']
 const ENC = new TextEncoder()
 const DEC = new TextDecoder()
 
@@ -65,6 +65,10 @@ export interface Target {
   dataset?: string
   agentId?: string
   facet?: string
+  // Where background-job consoles live: { type: 'redis' } puts each
+  // job's console on its own Redis stream (REDIS_URL). Only the ram
+  // opener consults it; main.ts refuses it on any other resource.
+  console?: { type?: string }
   clis?: string[]
   // Scope an installed account CLI to this mount's folder, so the CLI and
   // the mount are pointed at the same place.

integ/runners/typescript/main.ts

@@ -73,6 +73,12 @@ async function runTarget(
   report: Report | null,
   emit: EmitRow[] | null,
 ): Promise<void> {
+  // A console block is only wired into the ram opener; refusing it
+  // anywhere else keeps a silently RAM-consoled "redis console" target
+  // from reading as covered.
+  if (target.console !== undefined && target.mounts[0].resource !== 'ram') {
+    throw new Error(`target ${target.id}: console targets ride ram mounts`)
+  }
   const { ws, cleanup } = await ADAPTERS[target.mounts[0].resource](target)
   try {
     // A target's declared environment. A CLI whose spec reads a variable

integ/targets.json

@@ -320,6 +320,29 @@
         }
       }
     },
+    {
+      "id": "console-redis",
+      "hosts": [
+        "python",
+        "typescript-node"
+      ],
+      "service": "redis",
+      "console": {
+        "type": "redis"
+      },
+      "mounts": [
+        {
+          "path": "/data",
+          "resource": "ram",
+          "backend": "memory"
+        },
+        {
+          "path": "/data2",
+          "resource": "ram",
+          "backend": "memory"
+        }
+      ]
+    },
     {
       "id": "opfs",
       "hosts": [