feat: harden chat state by tenant

packages/casan-control-panel/README.md

@@ -69,9 +69,9 @@ Governed Chat (Plan-18 MVP-0/1/2):
   with role visibility; selected agent/skill/delegation are bound by harness
   `chat-agent-resolver.py`.
 - `GET /api/v1/chat/audit/verify` — verifies the chat audit hash chain.
-- `GET /api/v1/chat/replay?chatId=<id>` — verifies chat-chain integrity,
+- `GET /api/v1/chat/replay?chatId=<id>&tenant=<id>` — verifies chat-chain integrity,
   evidence artifact hashes, and OPERATOR loop trace replay through harness
-  `chat-replay.py`.
+  `chat-replay.py`; non-default tenants read only their tenant partition.
 - `GET /api/v1/command` — includes the `chat_loop` Command Center widget backed
   by chat audit/replay evidence, loop ticker rows, and token budget gauge.
 - Delegation escalation returns `ESCALATED` and creates a pending `chat.escalate`
@@ -79,6 +79,9 @@ Governed Chat (Plan-18 MVP-0/1/2):
 - CODEGEN requests are draft-only: `codegen-draft` outputs a state artifact, runs
   `artifact-scan.sh` + `tool-output-scan.sh`, certifies through Plan-17 loop-run,
   and never writes to the source tree.
+- Non-default tenant chat state is resolved through `tenant-store.sh`; replay path
+  overrides are guarded, audit snapshots are encrypted with `tenant-crypt.sh`, and
+  tenant kill-switch/quota checks run before chat work.
 - `/chat` UI shows actor/role scope, `mode/risk/decision` badges, certified answer,
   evidence sources, registered operator actions, agent binding, loop certification,
   codegen draft scan status, action-gate status, router details, and audit hash. Side-effect requests outside

packages/casan-control-panel/backend/src/chat/chat.controller.ts

@@ -18,8 +18,8 @@ export class ChatController {
   }
 
   @Get('replay')
-  replay(@Query('chatId') chatId?: string, @Query('turnId') turnId?: string) {
-    return ok(this.svc.replay(chatId || '', turnId || ''));
+  replay(@Query('chatId') chatId?: string, @Query('turnId') turnId?: string, @Query('tenant') tenant?: string) {
+    return ok(this.svc.replay(chatId || '', turnId || '', tenant || ''));
   }
 
   @Get('actions')

packages/casan-control-panel/backend/src/chat/chat.service.ts

@@ -25,13 +25,13 @@ const AGENT_CLI = join(HARNESS_BIN, 'chat-agent-resolver.py');
 const REPLAY_CLI = join(HARNESS_BIN, 'chat-replay.py');
 const RBAC_CLI = join(HARNESS_BIN, 'rbac-check.py');
 
-function runPython(script: string, args: string[]): CommandResult {
+function runPython(script: string, args: string[], extraEnv: NodeJS.ProcessEnv = {}): CommandResult {
   try {
     const stdout = execFileSync('python3', [script, ...args], {
       cwd: APP_ROOT,
       encoding: 'utf8',
       stdio: ['ignore', 'pipe', 'pipe'],
-      env: process.env,
+      env: { ...process.env, ...extraEnv },
     });
     return { status: 0, stdout: stdout.trim(), stderr: '' };
   } catch (err: any) {
@@ -94,11 +94,12 @@ export class ChatService {
     return { ok: res.status === 0, output: res.stdout || res.stderr };
   }
 
-  replay(chatId = '', turnId = '') {
+  replay(chatId = '', turnId = '', tenant = '') {
     const args = ['replay'];
     if (chatId) args.push('--chat-id', chatId);
     if (turnId) args.push('--turn-id', turnId);
-    const res = runPython(REPLAY_CLI, args);
+    const env = tenant && tenant !== 'default' ? { CASAN_TENANT_ID: tenant } : {};
+    const res = runPython(REPLAY_CLI, args, env);
     const parsed = parseJson<Record<string, any>>(res.stdout);
     if (parsed) return { ok: res.status === 0, ...parsed };
     throw new InternalServerErrorException(res.stderr || res.stdout || 'CHAT_REPLAY_FAILED');

packages/casan-control-panel/backend/test/chat.test.ts

@@ -15,9 +15,11 @@ function withTempChatState(fn: () => void) {
     CASAN_CHAT_AUDIT_HEAD: process.env.CASAN_CHAT_AUDIT_HEAD,
     CASAN_CHAT_METRICS_LOG: process.env.CASAN_CHAT_METRICS_LOG,
     CASAN_APPROVAL_INBOX_FILE: process.env.CASAN_APPROVAL_INBOX_FILE,
+    CASAN_TENANT_STATE_ROOT: process.env.CASAN_TENANT_STATE_ROOT,
   };
   const state = mkdtempSync(join(tmpdir(), 'cp-chat-'));
   process.env.CASAN_STATE_ROOT = state;
+  process.env.CASAN_TENANT_STATE_ROOT = join(state, 'tenants');
   process.env.CASAN_APPROVAL_INBOX_FILE = join(state, 'approval-inbox.json');
   delete process.env.CASAN_CHAT_AUDIT_LOG;
   delete process.env.CASAN_CHAT_AUDIT_HEAD;
@@ -135,3 +137,22 @@ test('chat ask escalates delegation approval into approval inbox', () => {
     assert.equal(inbox.proposals[0].payload.agent_binding.decision, 'REQUIRES_APPROVAL');
   });
 });
+
+test('chat replay is partitioned by non-default tenant', () => {
+  withTempChatState(() => {
+    const svc = new ChatService();
+    const alpha = { ...viewer, actor: 'tenant-alpha', tenant: 'alpha' };
+    const res = svc.ask({ message: 'Summarize Plan 18 MVP-2 status', chatId: 'tenant-chat' }, alpha) as any;
+    assert.equal(res.success, true);
+    assert.equal(res.audit_verify.ok, true);
+
+    const alphaReplay = svc.replay('tenant-chat', '', 'alpha') as any;
+    assert.equal(alphaReplay.ok, true);
+    assert.equal(alphaReplay.decision, 'MATCH');
+    assert.equal(alphaReplay.records, 1);
+
+    const betaReplay = svc.replay('tenant-chat', '', 'beta') as any;
+    assert.equal(betaReplay.ok, true);
+    assert.equal(betaReplay.records, 0);
+  });
+});

packages/casan-control-panel/frontend/src/lib/api.ts

@@ -220,7 +220,7 @@ export const api = {
   askChat: (actor: SettingsActor, body: { message: string; chatId?: string; agentId?: string; skillId?: string; delegationLevel?: number }) =>
     post<ChatAnswer>('chat/ask', body, actorHeaders(actor)),
   verifyChatAudit: () => get<{ ok: boolean; output: string }>('chat/audit/verify'),
-  replayChat: (chatId = '', turnId = '') => get<ChatReplay>(`chat/replay?chatId=${encodeURIComponent(chatId)}&turnId=${encodeURIComponent(turnId)}`),
+  replayChat: (chatId = '', turnId = '', tenant = '') => get<ChatReplay>(`chat/replay?chatId=${encodeURIComponent(chatId)}&turnId=${encodeURIComponent(turnId)}&tenant=${encodeURIComponent(tenant)}`),
   chatActions: (actor: SettingsActor) => getWithHeaders<{ success: boolean; actions: ChatAction[] }>('chat/actions', actorHeaders(actor)),
   chatAgents: (actor: SettingsActor) => getWithHeaders<{ success: boolean; agents: ChatAgent[] }>('chat/agents', actorHeaders(actor)),
 };

packages/casan-harness/scripts/bash/approval-inbox.py

@@ -13,6 +13,7 @@ import argparse
 import hashlib
 import json
 import os
+import subprocess
 import sys
 from datetime import datetime, timezone
 
@@ -31,6 +32,7 @@ def project_root() -> str:
 
 ROOT = project_root()
 HARNESS_ROOT = os.path.join(ROOT, "packages", "casan-harness")
+TENANT_STORE = os.path.join(HARNESS_ROOT, "scripts", "bash", "tenant-store.sh")
 
 
 def now_iso() -> str:
@@ -38,9 +40,26 @@ def now_iso() -> str:
 
 
 def inbox_path() -> str:
-    return os.environ.get("CASAN_APPROVAL_INBOX_FILE") or os.path.join(
-        ROOT, ".specify", "level5", "approval-inbox.json"
-    )
+    if os.environ.get("CASAN_APPROVAL_INBOX_FILE"):
+        return guarded_override(os.environ["CASAN_APPROVAL_INBOX_FILE"])
+    if os.environ.get("CASAN_TENANT_ID"):
+        return tenant_path("approvals/approval-inbox.json")
+    return os.path.join(ROOT, ".specify", "level5", "approval-inbox.json")
+
+
+def tenant_path(logical: str) -> str:
+    r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+    if r.returncode != 0:
+        raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return r.stdout.strip()
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
 
 
 def policy_path() -> str:

packages/casan-harness/scripts/bash/chat-agent-resolver.py

@@ -29,6 +29,7 @@ ROOT = project_root()
 HARNESS_ROOT = os.path.join(ROOT, "packages", "casan-harness")
 HARNESS_BIN = os.path.join(HARNESS_ROOT, "scripts", "bash")
 RBAC = os.path.join(HARNESS_BIN, "rbac-check.py")
+TENANT_STORE = os.path.join(HARNESS_BIN, "tenant-store.sh")
 GENESIS_HASH = "0" * 64
 
 
@@ -44,16 +45,33 @@ def state_root() -> str:
     return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
 
 
+def tenant_path(logical: str, fallback: str) -> str:
+    if os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+        return r.stdout.strip()
+    return os.path.join(state_root(), fallback)
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
+
+
 def registry_path() -> str:
     return os.environ.get("CASAN_AGENT_REGISTRY_FILE") or os.path.join(HARNESS_ROOT, "config", "agent-registry.yaml")
 
 
 def audit_path() -> str:
-    return os.environ.get("CASAN_AGENT_BIND_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "agent-bindings.jsonl")
+    return guarded_override(os.environ["CASAN_AGENT_BIND_AUDIT_LOG"]) if os.environ.get("CASAN_AGENT_BIND_AUDIT_LOG") else tenant_path("chat/agent-bindings.jsonl", "logs/chat/agent-bindings.jsonl")
 
 
 def head_path() -> str:
-    return os.environ.get("CASAN_AGENT_BIND_AUDIT_HEAD") or os.path.join(state_root(), "logs", "chat", "agent-bindings-head.txt")
+    return guarded_override(os.environ["CASAN_AGENT_BIND_AUDIT_HEAD"]) if os.environ.get("CASAN_AGENT_BIND_AUDIT_HEAD") else tenant_path("chat/agent-bindings-head.txt", "logs/chat/agent-bindings-head.txt")
 
 
 def load_registry():
@@ -149,6 +167,8 @@ def run_rbac(args):
 
 
 def bind(args) -> int:
+    if args.tenant and args.tenant != "default":
+        os.environ["CASAN_TENANT_ID"] = args.tenant
     registry = load_registry()
     if registry.get("_error"):
         out = decision_payload(args, "DENIED", registry["_error"])

packages/casan-harness/scripts/bash/chat-operator.py

@@ -35,30 +35,49 @@ HARNESS_BIN = os.path.join(HARNESS_ROOT, "scripts", "bash")
 ROUTER = os.path.join(HARNESS_BIN, "prompt-mode-router.py")
 SECURITY = os.path.join(HARNESS_BIN, "security-check.sh")
 ACTION_GATE = os.path.join(HARNESS_BIN, "action-gate.sh")
+TENANT_STORE = os.path.join(HARNESS_BIN, "tenant-store.sh")
+TENANT_CRYPT = os.path.join(HARNESS_BIN, "tenant-crypt.sh")
 
 
 def state_root() -> str:
     return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
 
 
+def tenant_path(logical: str, fallback: str) -> str:
+    if os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+        return r.stdout.strip()
+    return os.path.join(state_root(), fallback)
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
+
+
 def config_path() -> str:
     return os.environ.get("CASAN_OPERATOR_ACTIONS_FILE") or os.path.join(HARNESS_ROOT, "config", "operator-actions.yaml")
 
 
 def audit_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "chat-turns.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_LOG"]) if os.environ.get("CASAN_CHAT_AUDIT_LOG") else tenant_path("chat/chat-turns.jsonl", "logs/chat/chat-turns.jsonl")
 
 
 def head_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_HEAD") or os.path.join(state_root(), "logs", "chat", "chat-head.txt")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_HEAD"]) if os.environ.get("CASAN_CHAT_AUDIT_HEAD") else tenant_path("chat/chat-head.txt", "logs/chat/chat-head.txt")
 
 
 def metrics_path() -> str:
-    return os.environ.get("CASAN_CHAT_METRICS_LOG") or os.path.join(state_root(), "logs", "cost", "metrics.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_METRICS_LOG"]) if os.environ.get("CASAN_CHAT_METRICS_LOG") else tenant_path("telemetry/cost/metrics.jsonl", "logs/cost/metrics.jsonl")
 
 
 def artifact_dir() -> str:
-    return os.path.join(state_root(), "logs", "chat", "operator-artifacts")
+    return tenant_path("chat/operator-artifacts", "logs/chat/operator-artifacts")
 
 
 def now_iso() -> str:
@@ -181,9 +200,18 @@ def record_turn(base):
     os.makedirs(os.path.dirname(head_path()), exist_ok=True)
     with open(head_path(), "w", encoding="utf-8") as fh:
         fh.write(record_hash + "\n")
+    encrypt_chat_audit_snapshot(path)
     return rec
 
 
+def encrypt_chat_audit_snapshot(path: str):
+    if not os.environ.get("CASAN_TENANT_ID"):
+        return
+    if not os.path.isfile(path):
+        return
+    subprocess.run(["bash", TENANT_CRYPT, "encrypt", path, path + ".enc"], cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
 def record_metrics(trace_id: str, message: str, answer: str, status: str, latency_ms: int, action_id: str):
     input_tokens = len(message.split())
     output_tokens = len(answer.split())
@@ -287,6 +315,8 @@ def list_actions(args) -> int:
 
 
 def run(args) -> int:
+    if args.tenant and args.tenant != "default":
+        os.environ["CASAN_TENANT_ID"] = args.tenant
     started = datetime.now(timezone.utc)
     trace_id = str(uuid.uuid4())
     chat_id = args.chat_id or "chat-default"

packages/casan-harness/scripts/bash/chat-readonly.py

@@ -39,22 +39,41 @@ ROOT = project_root()
 HARNESS_BIN = os.path.join(ROOT, "packages", "casan-harness", "scripts", "bash")
 ROUTER = os.path.join(HARNESS_BIN, "prompt-mode-router.py")
 SECURITY = os.path.join(HARNESS_BIN, "security-check.sh")
+TENANT_STORE = os.path.join(HARNESS_BIN, "tenant-store.sh")
+TENANT_CRYPT = os.path.join(HARNESS_BIN, "tenant-crypt.sh")
 
 
 def state_root() -> str:
     return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
 
 
+def tenant_path(logical: str, fallback: str) -> str:
+    if os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+        return r.stdout.strip()
+    return os.path.join(state_root(), fallback)
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
+
+
 def audit_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "chat-turns.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_LOG"]) if os.environ.get("CASAN_CHAT_AUDIT_LOG") else tenant_path("chat/chat-turns.jsonl", "logs/chat/chat-turns.jsonl")
 
 
 def head_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_HEAD") or os.path.join(state_root(), "logs", "chat", "chat-head.txt")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_HEAD"]) if os.environ.get("CASAN_CHAT_AUDIT_HEAD") else tenant_path("chat/chat-head.txt", "logs/chat/chat-head.txt")
 
 
 def metrics_path() -> str:
-    return os.environ.get("CASAN_CHAT_METRICS_LOG") or os.path.join(state_root(), "logs", "cost", "metrics.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_METRICS_LOG"]) if os.environ.get("CASAN_CHAT_METRICS_LOG") else tenant_path("telemetry/cost/metrics.jsonl", "logs/cost/metrics.jsonl")
 
 
 def now_iso() -> str:
@@ -223,9 +242,18 @@ def record_turn(base):
     os.makedirs(os.path.dirname(head_path()), exist_ok=True)
     with open(head_path(), "w", encoding="utf-8") as fh:
         fh.write(record_hash + "\n")
+    encrypt_chat_audit_snapshot(path)
     return rec
 
 
+def encrypt_chat_audit_snapshot(path: str):
+    if not os.environ.get("CASAN_TENANT_ID"):
+        return
+    if not os.path.isfile(path):
+        return
+    subprocess.run(["bash", TENANT_CRYPT, "encrypt", path, path + ".enc"], cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
 def record_metrics(trace_id: str, message: str, answer: str, status: str, latency_ms: int):
     input_tokens = len(message.split())
     output_tokens = len(answer.split())
@@ -253,6 +281,8 @@ def record_metrics(trace_id: str, message: str, answer: str, status: str, latenc
 
 
 def ask(args):
+    if args.tenant and args.tenant != "default":
+        os.environ["CASAN_TENANT_ID"] = args.tenant
     started = datetime.now(timezone.utc)
     trace_id = str(uuid.uuid4())
     message = args.message

packages/casan-harness/scripts/bash/chat-replay.py

@@ -33,14 +33,32 @@ def project_root() -> str:
 ROOT = project_root()
 BIN = os.path.join(ROOT, "packages", "casan-harness", "scripts", "bash")
 LOOP_TRACE = os.path.join(BIN, "loop-trace.py")
+TENANT_STORE = os.path.join(BIN, "tenant-store.sh")
 
 
 def state_root() -> str:
     return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
 
 
+def tenant_path(logical: str, fallback: str) -> str:
+    if os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+        return r.stdout.strip()
+    return os.path.join(state_root(), fallback)
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
+
+
 def audit_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "chat-turns.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_LOG"]) if os.environ.get("CASAN_CHAT_AUDIT_LOG") else tenant_path("chat/chat-turns.jsonl", "logs/chat/chat-turns.jsonl")
 
 
 def sha_text(text: str) -> str:
@@ -87,12 +105,12 @@ def resolve_path(path: str) -> str:
 def loop_state_root(loop_run):
     explicit = os.environ.get("CASAN_LOOP_STATE_ROOT")
     if explicit:
-        return explicit
+        return guarded_override(explicit)
     artifact = loop_run.get("artifact") or ""
     marker = os.sep + "logs" + os.sep + "chat" + os.sep + "loop-runs" + os.sep
     if marker in artifact:
         return artifact.split(marker, 1)[0] + os.sep + "logs" + os.sep + "chat" + os.sep + "loop-state"
-    return os.path.join(state_root(), "logs", "chat", "loop-state")
+    return tenant_path("chat/loop-state", "logs/chat/loop-state")
 
 
 def run_loop_trace(loop_run, cmd: str):

packages/casan-harness/scripts/bash/chat-turn.py

@@ -38,22 +38,43 @@ PREFLIGHT = os.path.join(BIN, "harness-preflight.sh")
 CONTEXT_SCAN = os.path.join(BIN, "context-assemble-scan.sh")
 TOOL_OUTPUT_SCAN = os.path.join(BIN, "tool-output-scan.sh")
 ARTIFACT_SCAN = os.path.join(BIN, "artifact-scan.sh")
+TENANT_STORE = os.path.join(BIN, "tenant-store.sh")
+TENANT_CRYPT = os.path.join(BIN, "tenant-crypt.sh")
+KILL_SWITCH = os.path.join(BIN, "kill-switch.sh")
+COST_SPIKE = os.path.join(BIN, "cost-spike-detect.sh")
 
 
 def state_root() -> str:
     return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
 
 
+def tenant_path(logical: str, fallback: str) -> str:
+    if os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+        return r.stdout.strip()
+    return os.path.join(state_root(), fallback)
+
+
+def guarded_override(path: str) -> str:
+    if path and os.environ.get("CASAN_TENANT_ID"):
+        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
+        if r.returncode != 0:
+            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
+    return path
+
+
 def now_iso() -> str:
     return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
 
 def audit_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "chat-turns.jsonl")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_LOG"]) if os.environ.get("CASAN_CHAT_AUDIT_LOG") else tenant_path("chat/chat-turns.jsonl", "logs/chat/chat-turns.jsonl")
 
 
 def head_path() -> str:
-    return os.environ.get("CASAN_CHAT_AUDIT_HEAD") or os.path.join(state_root(), "logs", "chat", "chat-head.txt")
+    return guarded_override(os.environ["CASAN_CHAT_AUDIT_HEAD"]) if os.environ.get("CASAN_CHAT_AUDIT_HEAD") else tenant_path("chat/chat-head.txt", "logs/chat/chat-head.txt")
 
 
 def sha(text: str) -> str:
@@ -147,11 +168,11 @@ def write_text(path: str, text: str):
 
 
 def loop_dir(run_id: str) -> str:
-    return os.path.join(state_root(), "logs", "chat", "loop-runs", run_id)
+    return tenant_path(f"chat/loop-runs/{run_id}", f"logs/chat/loop-runs/{run_id}")
 
 
 def codegen_dir(run_id: str) -> str:
-    return os.path.join(state_root(), "logs", "chat", "codegen-artifacts", run_id)
+    return tenant_path(f"chat/codegen-artifacts/{run_id}", f"logs/chat/codegen-artifacts/{run_id}")
 
 
 def load_chat_head() -> str:
@@ -177,9 +198,65 @@ def append_chat_turn(base):
     os.makedirs(os.path.dirname(head_path()), exist_ok=True)
     with open(head_path(), "w", encoding="utf-8") as fh:
         fh.write(record_hash + "\n")
+    encrypt_chat_audit_snapshot(path)
     return rec
 
 
+def encrypt_chat_audit_snapshot(path: str):
+    if not os.environ.get("CASAN_TENANT_ID"):
+        return
+    if not os.path.isfile(path):
+        return
+    enc = path + ".enc"
+    subprocess.run(["bash", TENANT_CRYPT, "encrypt", path, enc], cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
+def tenant_runtime_guard(args):
+    if args.tenant and args.tenant != "default":
+        os.environ["CASAN_TENANT_ID"] = args.tenant
+    if os.environ.get("CASAN_TENANT_ID"):
+        ks = subprocess.run(["bash", KILL_SWITCH, "check", "tenant", os.environ["CASAN_TENANT_ID"]], cwd=ROOT, capture_output=True, text=True)
+        if ks.returncode != 0:
+            print(json.dumps({
+                "success": False,
+                "mode": "BLOCK",
+                "risk": "high",
+                "decision": "HALTED",
+                "answer": f"Tenant kill-switch active for {os.environ['CASAN_TENANT_ID']}.",
+                "sources": [],
+                "certified": False,
+                "audit": {},
+                "router": {"reason": "tenant_kill_switch_active", "matched_rules": ["tenant_kill_switch"]},
+            }, ensure_ascii=False))
+            return 3
+        if os.environ.get("CASAN_COST_CUMULATIVE_BUDGET_TOKENS"):
+            quota = subprocess.run(["bash", COST_SPIKE], cwd=ROOT, capture_output=True, text=True)
+            if quota.returncode == 2:
+                print(json.dumps({
+                    "success": False,
+                    "mode": "BLOCK",
+                    "risk": "high",
+                    "decision": "HALTED",
+                    "answer": f"Tenant quota exceeded for {os.environ['CASAN_TENANT_ID']}.",
+                    "sources": [],
+                    "certified": False,
+                    "audit": {},
+                    "router": {"reason": "tenant_quota_exceeded", "matched_rules": ["tenant_quota"]},
+                }, ensure_ascii=False))
+                return 3
+    return 0
+
+
+def loop_env_for(args):
+    env = {
+        **os.environ,
+        "CASAN_LOOP_STATE_ROOT": os.environ.get("CASAN_LOOP_STATE_ROOT") or tenant_path("chat/loop-state", "logs/chat/loop-state"),
+    }
+    if os.environ.get("CASAN_TENANT_ID"):
+        env["CASAN_TENANT_ID"] = os.environ["CASAN_TENANT_ID"]
+    return env
+
+
 def run_preflight_and_context(run_id: str, draft_path: str):
     preflight_out = os.path.join(loop_dir(run_id), "preflight.json")
     r = subprocess.run(["bash", PREFLIGHT, draft_path, preflight_out, "--model", "local:chat-turn"], cwd=ROOT, capture_output=True, text=True)
@@ -235,11 +312,7 @@ def certify_operator_draft(args, router, binding):
             "success_criteria": criteria_path,
         }, 2
 
-    loop_env = {
-        **os.environ,
-        "CASAN_LOOP_STATE_ROOT": os.environ.get("CASAN_LOOP_STATE_ROOT") or os.path.join(state_root(), "logs", "chat", "loop-state"),
-        "CASAN_TENANT_ID": args.tenant,
-    }
+    loop_env = loop_env_for(args)
     dlevel = f"L{binding.get('delegation_level', 0)}"
     r = subprocess.run([
         "bash", LOOP_RUN,
@@ -318,11 +391,7 @@ def certify_codegen_draft(args, router, binding):
             "artifact_scan": {"ok": False, "output": scan_out},
         }, 2
 
-    loop_env = {
-        **os.environ,
-        "CASAN_LOOP_STATE_ROOT": os.environ.get("CASAN_LOOP_STATE_ROOT") or os.path.join(state_root(), "logs", "chat", "loop-state"),
-        "CASAN_TENANT_ID": args.tenant,
-    }
+    loop_env = loop_env_for(args)
     dlevel = f"L{binding.get('delegation_level', 0)}"
     r = subprocess.run([
         "bash", LOOP_RUN,
@@ -530,6 +599,9 @@ def submit_escalation(args, router, binding):
 
 
 def ask(args) -> int:
+    guard_rc = tenant_runtime_guard(args)
+    if guard_rc != 0:
+        return guard_rc
     router = classify(args.message)
     bind_rc, binding = bind_agent(args, router)
     if bind_rc != 0:

packages/casan-harness/scripts/bash/ci-harness-gate.sh

@@ -155,6 +155,7 @@ run "phase-chat-stream-hold"     bash "$TESTS/phase-chat-stream-hold-tests.sh"
 run "phase-chat-replay"          bash "$TESTS/phase-chat-replay-tests.sh"
 run "phase-chat-approval"        bash "$TESTS/phase-chat-approval-tests.sh"
 run "phase-chat-codegen"         bash "$TESTS/phase-chat-codegen-tests.sh"
+run "phase-chat-tenant"          bash "$TESTS/phase-chat-tenant-tests.sh"
 
 # ARCH-02: coverage cannot silently drop; ARCH-01: harness/policy cannot silently
 # drift. Both SKIP cleanly when no manifest is provisioned (non-strict dev/CI).

packages/casan-harness/tests/phase-chat-approval-tests.sh

@@ -26,7 +26,7 @@ python3 "$TURN" ask \
   --actor alice \
   --role viewer \
   --chat-id approval-chat \
-  --tenant tenant-a \
+  --tenant default \
   --delegation-level 1 > "$WORK/escalated.json"
 RC=$?
 set -e 2>/dev/null || true

packages/casan-harness/tests/phase-chat-tenant-tests.sh

@@ -0,0 +1,101 @@
+#!/usr/bin/env bash
+set -uo pipefail
+
+# Plan-18 MVP-3 Track 9: chat state is tenant-partitioned, guarded,
+# encrypted-at-rest as an offline snapshot, and scoped for tenant resources.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "$SCRIPT_DIR/../scripts/bash/casan-paths.sh"
+S="$CASAN_HARNESS_ROOT/scripts/bash"
+TURN="$S/chat-turn.py"
+REPLAY="$S/chat-replay.py"
+TS="$S/tenant-store.sh"
+TC="$S/tenant-crypt.sh"
+KS="$S/kill-switch.sh"
+WORK="$(mktemp -d)"
+trap 'rm -rf "$WORK"' EXIT
+export CASAN_STATE_ROOT="$WORK/state"
+export CASAN_TENANT_STATE_ROOT="$WORK/tenants"
+export CASAN_KILLSWITCH_DIR="$WORK/ks"
+
+PASS=0; FAIL=0
+pass() { echo "PASS: $1"; PASS=$((PASS + 1)); }
+fail() { echo "FAIL: $1"; FAIL=$((FAIL + 1)); }
+rc_of() { set +e; "$@" >/dev/null 2>&1; echo $?; set -e 2>/dev/null || true; }
+
+echo "===== Plan-18 MVP-3 chat tenant hardening ====="
+
+CASAN_TENANT_ID=alpha python3 "$TURN" ask --message "Summarize Plan 18 status" --actor alice --role viewer --tenant alpha --chat-id tenant-chat > "$WORK/alpha.json"
+CASAN_TENANT_ID=beta python3 "$TURN" ask --message "Summarize Plan 18 status" --actor bob --role viewer --tenant beta --chat-id tenant-chat > "$WORK/beta.json"
+
+pA="$(CASAN_TENANT_ID=alpha bash "$TS" resolve chat/chat-turns.jsonl 2>/dev/null)"
+pB="$(CASAN_TENANT_ID=beta bash "$TS" resolve chat/chat-turns.jsonl 2>/dev/null)"
+{ [[ -f "$pA" && -f "$pB" && "$pA" != "$pB" ]]; } \
+  && pass "18.9.1 tenant A/B chat audits are separate files" || fail "chat audit partitioning failed"
+
+CASAN_TENANT_ID=alpha python3 "$REPLAY" verify-chain > "$WORK/alpha-chain.json"
+CASAN_TENANT_ID=beta python3 "$REPLAY" verify-chain > "$WORK/beta-chain.json"
+python3 - "$WORK/alpha-chain.json" "$WORK/beta-chain.json" <<'PY' \
+  && pass "18.9.1 each tenant verifies only its own chat chain" || fail "tenant chat chain verify mismatch"
+import json, sys
+a = json.load(open(sys.argv[1]))
+b = json.load(open(sys.argv[2]))
+assert a["decision"] == "OK" and a["records"] == 1
+assert b["decision"] == "OK" and b["records"] == 1
+PY
+
+[[ "$(rc_of env CASAN_TENANT_ID=alpha CASAN_CHAT_AUDIT_LOG="$pB" python3 "$REPLAY" verify-chain)" -ne 0 ]] \
+  && pass "18.9.1 explicit cross-tenant audit path is denied" || fail "cross-tenant audit override was allowed"
+
+[[ -f "$pA.enc" ]] \
+  && pass "18.9.2 encrypted chat audit snapshot is written" || fail "encrypted audit snapshot missing"
+CASAN_TENANT_ID=alpha bash "$TC" decrypt "$pA.enc" "$WORK/alpha.dec" >/dev/null 2>&1
+grep -q "tenant-chat" "$WORK/alpha.dec" \
+  && pass "18.9.2 tenant A decrypts its own chat audit snapshot" || fail "tenant A could not decrypt own audit"
+CASAN_TENANT_ID=beta bash "$TC" decrypt "$pA.enc" "$WORK/beta.dec" >/dev/null 2>&1 || true
+grep -q "tenant-chat" "$WORK/beta.dec" \
+  && fail "18.9.2 tenant B read tenant A audit plaintext" \
+  || pass "18.9.2 tenant B cannot decrypt tenant A audit snapshot"
+
+bash "$KS" engage tenant alpha "chat incident" >/dev/null 2>&1
+set +e
+CASAN_TENANT_ID=alpha python3 "$TURN" ask --message "Summarize Plan 18 status" --actor alice --role viewer --tenant alpha --chat-id killed > "$WORK/killed.json"
+RC_A=$?
+CASAN_TENANT_ID=beta python3 "$TURN" ask --message "Summarize Plan 18 status" --actor bob --role viewer --tenant beta --chat-id alive > "$WORK/alive.json"
+RC_B=$?
+set -e 2>/dev/null || true
+python3 - "$WORK/killed.json" "$WORK/alive.json" "$RC_A" "$RC_B" <<'PY' \
+  && pass "18.9.3 tenant kill-switch halts only its own tenant" || fail "tenant kill-switch isolation failed"
+import json, sys
+k = json.load(open(sys.argv[1]))
+a = json.load(open(sys.argv[2]))
+assert int(sys.argv[3]) == 3
+assert k["decision"] == "HALTED"
+assert int(sys.argv[4]) == 0
+assert a["success"] is True
+PY
+
+gLOG="$(CASAN_TENANT_ID=gamma bash "$TS" resolve telemetry/provider-usage.jsonl 2>/dev/null)"
+dLOG="$(CASAN_TENANT_ID=delta bash "$TS" resolve telemetry/provider-usage.jsonl 2>/dev/null)"
+printf '{"step":"chat","total_tokens":9000}\n' > "$gLOG"
+printf '{"step":"chat","total_tokens":10}\n' > "$dLOG"
+set +e
+CASAN_TENANT_ID=gamma CASAN_COST_CUMULATIVE_BUDGET_TOKENS=1000 python3 "$TURN" ask --message "Summarize Plan 18 status" --actor g --role viewer --tenant gamma --chat-id quota > "$WORK/quota.json"
+RC_G=$?
+CASAN_TENANT_ID=delta CASAN_COST_CUMULATIVE_BUDGET_TOKENS=1000 python3 "$TURN" ask --message "Summarize Plan 18 status" --actor d --role viewer --tenant delta --chat-id quota > "$WORK/quota-ok.json"
+RC_D=$?
+set -e 2>/dev/null || true
+python3 - "$WORK/quota.json" "$WORK/quota-ok.json" "$RC_G" "$RC_D" <<'PY' \
+  && pass "18.9.3 tenant quota halts only over-budget tenant" || fail "tenant quota isolation failed"
+import json, sys
+q = json.load(open(sys.argv[1]))
+ok = json.load(open(sys.argv[2]))
+assert int(sys.argv[3]) == 3
+assert q["decision"] == "HALTED"
+assert int(sys.argv[4]) == 0
+assert ok["success"] is True
+PY
+
+echo ""
+echo "===== CHAT TENANT SUMMARY: PASS=$PASS FAIL=$FAIL ====="
+[[ "$FAIL" -eq 0 ]] || exit 1

packages/casan-platform/README.md

@@ -19,7 +19,7 @@ Optional layer for teams that want UI / dashboard / visibility. Packages: `casan
 | RBAC + approval inbox | ✅ local-prod done | Settings + kill-switch API RBAC enforcement, approval inbox/delegation/oversight, SoD, governed setting proposal apply, and local OIDC claim→role mapping smoke are done. Enterprise IdP rollout remains production follow-up. |
 | Evidence Pack Viewer | 📋 planned | reads `docs/output/casan/evidence-packs/` |
 | Attack Battery Viewer | 📋 planned | reads red-team corpus + H4 recall results |
-| Read-only Ask CASAN + Operator/CODEGEN + agent selection + loop-hold + replay/widget/approvals | 🟡 MVP-0/1/2 done+test; MVP-3 planned | Plan-18: `POST /api/v1/chat/ask`, `GET /api/v1/chat/actions`, `GET /api/v1/chat/agents`, `GET /api/v1/chat/replay`, Command Center `chat_loop` widget, chat escalation into `/approvals`, `/chat` + `/command`, backed by harness `chat-turn.py`, `chat-agent-resolver.py`, `chat-replay.py`, `approval-inbox.py`, `artifact-scan.sh`, and Plan-17 `loop-run.sh` |
+| Read-only Ask CASAN + Operator/CODEGEN + agent selection + loop-hold + replay/widget/approvals + tenant hardening | ✅ MVP-0/1/2/3 done+test | Plan-18: `POST /api/v1/chat/ask`, `GET /api/v1/chat/actions`, `GET /api/v1/chat/agents`, `GET /api/v1/chat/replay`, Command Center `chat_loop` widget, chat escalation into `/approvals`, tenant-scoped replay, `/chat` + `/command`, backed by harness `chat-turn.py`, `chat-agent-resolver.py`, `chat-replay.py`, `approval-inbox.py`, `artifact-scan.sh`, `tenant-store.sh`, `tenant-crypt.sh`, and Plan-17 `loop-run.sh` |
 | Gitea webhook integration | 📋 planned | trigger gate / publish evidence on push |
 
 ## Build (preview)
@@ -29,6 +29,6 @@ scripts/package-release.sh platform     # → dist/casan-platform-preview-vX.Y.Z
 The bundle includes a `PREVIEW-INCOMPLETE.txt` marker. Do not treat it as a finished product.
 
 ## To implement later
-Start from Plan-15 RAI view or Plan-18 Track 9 multi-tenant hardening. Keep new numbers
+Start from Plan-15 RAI view or the next backlog item. Keep new numbers
 evidence-backed with provenance; any write/governed action must continue to route through
 harness RBAC, approval, and audit primitives.