Add approved parallel analysis, model routing and workspace artifact links

packages/dsh-multi-agent/README.md

@@ -72,6 +72,12 @@ strict JSON Schema validation with duplicate keys, non-finite constants and
 external schema retrieval rejected. Existing domain wrappers can retain their
 own parser and evidence/version checks. Usage is shared, not reimplemented.
 
+Python hosts can use `RoutingPolicy.select()` with trusted low-risk task facts and
+input byte size to choose a configured fast/strong model. This pure policy returns
+an auditable decision and fingerprint; it never calls a model, changes permissions
+or retries. Domain classification and applying the decision belong to the caller.
+The `ctx.multiAgent` JavaScript task registry still uses its configured model.
+
 ## Boundaries
 
 Harness already provides `ctx.subagents`, `subagent-spawn-in-process`, and

packages/dsh-multi-agent/python/dsh_multi_agent/__init__.py

@@ -3,7 +3,8 @@
 from .runner import RunCancelled, TaskDefinition, TaskRun, TaskRunner, normalize_result
 from .structured import json_result
 from .usage import RunUsage, normalize_usage, summarize
+from .routing import RoutingPolicy
 
 __all__ = ["HarnessRunResult", "IsolatedHarnessBackend", "Settings", "RunCancelled",
            "TaskDefinition", "TaskRun", "TaskRunner", "normalize_result", "json_result",
-           "RunUsage", "normalize_usage", "summarize"]
+           "RunUsage", "normalize_usage", "summarize", "RoutingPolicy"]

packages/dsh-multi-agent/python/dsh_multi_agent/parallel.py

@@ -0,0 +1,59 @@
+"""Bounded fan-out; authorization and task-specific execution belong to callers."""
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from contextlib import contextmanager
+import fcntl
+import os
+from pathlib import Path
+import tempfile
+
+
+@contextmanager
+def local_replica_slot():
+    """Four replicas per OS user across local workers; not a distributed quota."""
+    folder = Path(tempfile.gettempdir()) / f"dsh-replicas-{os.getuid()}"
+    folder.mkdir(mode=0o700, exist_ok=True)
+    if folder.is_symlink() or folder.stat().st_uid != os.getuid():
+        raise PermissionError("Unsafe replica quota directory")
+    acquired = None
+    for i in range(4):
+        handle = (folder / f"slot-{i}").open("a")
+        try:
+            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            acquired = handle
+            break
+        except BlockingIOError:
+            handle.close()
+    if acquired is None:
+        raise RuntimeError("Local replica capacity exhausted; no model call made")
+    try:
+        yield
+    finally:
+        fcntl.flock(acquired, fcntl.LOCK_UN)
+        acquired.close()
+
+
+def run_parallel(tasks, run, *, cancelled=lambda: False, on_result=lambda row: None,
+                 max_parallel=2):
+    if not 1 <= max_parallel <= 2 or not 1 <= len(tasks) <= 4:
+        raise ValueError("At most four tasks and two concurrent replicas are allowed")
+    identifiers = [task["task_id"] for task in tasks]
+    if any(not isinstance(key, str) or not key for key in identifiers) or len(set(identifiers)) != len(identifiers):
+        raise ValueError("Replica task IDs must be nonempty and unique")
+
+    def execute(task):
+        if cancelled():
+            return {"task_id": task["task_id"], "status": "cancelled"}
+        try:
+            return {**run(task, cancelled), "task_id": task["task_id"]}
+        except Exception as error:
+            # Arbitrary provider errors can contain secrets. Retain the category only.
+            return {"task_id": task["task_id"], "status": "failed", "error_type": type(error).__name__}
+
+    results = {}
+    with ThreadPoolExecutor(max_workers=max_parallel) as pool:
+        futures = [pool.submit(execute, task) for task in tasks]
+        for future in as_completed(futures):
+            row = future.result()
+            results[row["task_id"]] = row
+            on_result(row)
+    return [results[task["task_id"]] for task in tasks]

packages/dsh-multi-agent/python/dsh_multi_agent/routing.py

@@ -0,0 +1,49 @@
+"""Deterministic model selection from host-supplied task facts, not model output."""
+from dataclasses import asdict, dataclass
+import hashlib
+import json
+
+
+@dataclass(frozen=True)
+class RoutingPolicy:
+    enabled: bool = False
+    fastModel: str = ""
+    strongModel: str = ""
+    fastMaxInputBytes: int = 12000
+    fastTimeoutSeconds: int = 120
+    strongTimeoutSeconds: int = 300
+
+    def __post_init__(self):
+        if type(self.enabled) is not bool:
+            raise ValueError("modelRouting.enabled must be boolean")
+        for name in ("fastModel", "strongModel"):
+            value = getattr(self, name)
+            if not isinstance(value, str) or len(value) > 256 or any(ord(c) < 32 for c in value):
+                raise ValueError("modelRouting model must be a single-line model ID")
+            if value != value.strip():
+                raise ValueError("modelRouting model cannot have surrounding whitespace")
+        for name, low, high in (("fastMaxInputBytes", 256, 64000),
+                                ("fastTimeoutSeconds", 30, 300), ("strongTimeoutSeconds", 30, 300)):
+            value = getattr(self, name)
+            if type(value) is not int or not low <= value <= high:
+                raise ValueError("Invalid modelRouting." + name)
+
+    def select(self, *, provider, default_model, input_bytes, low_risk=False, reason="unknown_task"):
+        if type(low_risk) is not bool:
+            raise ValueError("low_risk must be a host boolean")
+        if type(input_bytes) is not int or input_bytes < 0:
+            raise ValueError("input_bytes must be a non-negative integer")
+        identity = {"version": 1, **asdict(self), "provider": provider, "default_model": default_model}
+        fingerprint = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest()
+        tier, model, why, timeout = "default", default_model, "routing_disabled", self.strongTimeoutSeconds
+        if self.enabled:
+            tier, model, why = "strong", self.strongModel or default_model, reason
+            if low_risk:
+                if input_bytes > self.fastMaxInputBytes:
+                    why = "input_exceeds_fast_budget"
+                elif not self.fastModel:
+                    why = "fast_model_not_configured"
+                else:
+                    tier, model, timeout = "fast", self.fastModel, self.fastTimeoutSeconds
+        return dict(tier=tier, model=model, provider=provider, reason=why,
+                    timeout_seconds=timeout, input_bytes=input_bytes, policy_fingerprint=fingerprint)

packages/dsh-multi-agent/tests/unit/parallel.spec.ts

@@ -0,0 +1,45 @@
+import { expect, it } from 'vitest'
+import { probe } from '../probe'
+
+it('caps fan-out and host-local admission; cancellation does not start pending children', () => {
+  const data = probe(`
+import json, tempfile, threading, time
+from contextlib import ExitStack
+from pathlib import Path
+from unittest.mock import patch
+from dsh_multi_agent.parallel import run_parallel, local_replica_slot
+out = {}
+tasks = [{"task_id":str(i)} for i in range(4)]
+def fail(fn):
+    try: fn(); return None
+    except Exception as e: return type(e).__name__
+out["too_many"] = fail(lambda: run_parallel(tasks * 2, lambda task, stop: {}))
+out["parallel_limit"] = fail(lambda: run_parallel(tasks, lambda task, stop: {}, max_parallel=3))
+with tempfile.TemporaryDirectory(prefix="dsh-quota-test-") as folder:
+    with patch("dsh_multi_agent.parallel.tempfile.gettempdir", return_value=folder):
+        with ExitStack() as slots:
+            for _ in range(4): slots.enter_context(local_replica_slot())
+            def fifth():
+                with local_replica_slot(): pass
+            out["overload"] = fail(fifth)
+        with local_replica_slot(): out["released"] = True
+out["quota_cleaned"] = not Path(folder).exists()
+stop = threading.Event()
+calls = []
+def run(task, cancelled):
+    calls.append(task["task_id"])
+    stop.set()
+    return {"status":"cancelled"}
+out["rows"] = run_parallel(tasks, run, cancelled=stop.is_set)
+out["calls"] = len(calls)
+print(json.dumps(out))
+`)
+  expect(data.too_many).toBe('ValueError')
+  expect(data.parallel_limit).toBe('ValueError')
+  expect(data.overload).toBe('RuntimeError')
+  expect(data.released).toBe(true)
+  expect(data.quota_cleaned).toBe(true)
+  expect(data.calls).toBeLessThanOrEqual(2)
+  expect(data.rows).toHaveLength(4)
+  expect(data.rows.every(row => row.status === 'cancelled')).toBe(true)
+})

packages/dsh-multi-agent/tests/unit/routing.spec.ts

@@ -0,0 +1,52 @@
+import { expect, it } from 'vitest'
+import { probe } from '../probe'
+
+const driver = `
+import json, sys
+from dsh_multi_agent.routing import RoutingPolicy
+data = json.load(sys.stdin)
+try:
+    result = RoutingPolicy(**data.get("policy", {})).select(provider="fixture", default_model="current",
+        input_bytes=data.get("bytes", 100), low_risk=data.get("low", False), reason="host_facts")
+except (ValueError, TypeError):
+    result = {"rejected":True}
+print(json.dumps(result))
+`
+
+it('keeps the current model when disabled, even with other models configured', () => {
+  expect(probe(driver, { policy: { fastModel: 'fast', strongModel: 'strong' }, low: true })).toMatchObject({
+    tier: 'default',
+    model: 'current',
+    reason: 'routing_disabled',
+  })
+})
+it('routes bounded low-risk facts, unknown tasks and oversized inputs deterministically', () => {
+  const policy = { enabled: true, fastModel: 'fast', strongModel: 'strong' }
+  expect(probe(driver, { policy, low: true })).toMatchObject({ tier: 'fast', model: 'fast', timeout_seconds: 120 })
+  expect(probe(driver, { policy })).toMatchObject({ tier: 'strong', model: 'strong', timeout_seconds: 300 })
+  expect(probe(driver, { policy, low: true, bytes: 12001 })).toMatchObject({
+    tier: 'strong',
+    reason: 'input_exceeds_fast_budget',
+  })
+  expect(probe(driver, { policy: { enabled: true }, low: true })).toMatchObject({
+    model: 'current',
+    reason: 'fast_model_not_configured',
+  })
+})
+it('fingerprints model IDs, time limits and input policy without including credentials', () => {
+  const one = probe(driver, { policy: { enabled: true } })
+  const two = probe(driver, { policy: { enabled: true, strongTimeoutSeconds: 250 } })
+  expect(one.policy_fingerprint).toMatch(/^[a-f0-9]{64}$/)
+  expect(one.policy_fingerprint).not.toBe(two.policy_fingerprint)
+})
+it.each([
+  { enabled: 'yes' },
+  { fastTimeoutSeconds: 301 },
+  { strongTimeoutSeconds: 29 },
+  { fastMaxInputBytes: true },
+  { fastModel: 'bad\nmodel' },
+  { strongModel: ' spaced ' },
+  { unknown: 1 },
+])('rejects invalid host configuration: %j', policy => {
+  expect(probe(driver, { policy })).toEqual({ rejected: true })
+})

packages/dsh-triton-riscv/.gitignore

@@ -7,6 +7,7 @@ python/*.egg-info/
 python/codex_agent/frontend/
 python/codex_agent/harness/bundle/
 __pycache__/
+.pytest_cache/
 *.pyc
 .state/
 .runtime/
@@ -19,3 +20,4 @@ __pycache__/
 !lib/config.js
 !lib/prompts.js
 !lib/tool-policy.js
+!lib/parallel-adapter.js

packages/dsh-triton-riscv/README.md

@@ -70,6 +70,43 @@ Validate relu_and_mul. Show the plan and run it after my approval.
 
 For remote tests, set `remote.host` and `remote.repository`; the server needs SSH access and the Triton/Buddy/LLVM toolchain.
 
+### Parallel Analysis
+
+For complex independent investigations, the assistant can propose temporary replicas.
+Review their goals, context and model in the approval dialog/card before they start.
+Structured multi-agent workflows also support this optional step. Declining it
+resumes the normal sequential workflow; approving it adds read-only findings before implementation.
+Each replica runs its own Harness loop with read-only source and retrieval tools:
+2 concurrent replicas, up to 4 subtasks, 8 tool calls and at most 300 seconds per replica.
+Writes and tests still require separate lifecycle approval. Results are findings, not test passes.
+Replica calls use the configured Responses model route (`DSH_PROVIDER`, `DSH_MODEL`,
+`DSH_API_KEY_ENV`, `ISRC_BASE_URL`); native sessions must match that provider.
+The local capacity guard allows 4 replicas per OS user, not a cluster-wide quota.
+
+### Optional Model Routing
+
+Add to the plugin's `config` (or the workbench's versioned `TRITON_RISCV_CONFIG`):
+
+```yaml
+modelRouting:
+  enabled: true
+  fastModel: YOUR_FAST_MODEL_ID
+  strongModel: YOUR_STRONG_MODEL_ID
+  fastMaxInputBytes: 12000
+  fastTimeoutSeconds: 120
+  strongTimeoutSeconds: 300
+```
+
+Both IDs must be available through the current provider, endpoint and API key.
+Bounded information tasks may use `fast`; code, review and unclear tasks use `strong`.
+Missing `fastModel` uses the strong/current model. Routing is off by default.
+The byte limit is a conservative routing gate, not a token count or a context-window guarantee.
+Applies to managed workbench turns, proposal roles and approved replicas; native main-chat
+and unmanaged legacy sessions keep their host model. Child models are shown before approval;
+changed routing settings require a new plan. Errors do not trigger an extra paid retry.
+Role/child deadlines use the values above; parent turns retain their existing outer
+deadline so they can await child batches or validation tools.
+
 ### Optional Workbench
 
 After installation, run from the repository root:
@@ -89,6 +126,8 @@ Live requests need separate `ISRC_API_KEY` and `DSH_MODEL` settings. Workbench a
 For invite-only team access, use the [account gateway](python/codex_agent/README.md#account-gateway)
 instead of exposing the single-user workbench. It serves the same frontend with
 registration, login and private per-account workspaces. Native Harness login is separate.
+Passwords require 6 to 128 characters. This is a trusted-workbench policy, not a
+recommendation for public deployment; use strong, unique passwords.
 
 ### Optional Redis Cache
 

packages/dsh-triton-riscv/frontend/src/AccountGate.test.tsx

@@ -35,7 +35,7 @@ it('blocks the workbench until login, then unmounts private data on logout', asy
   expect(screen.queryByText('private workspace')).not.toBeInTheDocument()
   await screen.findByRole('heading', { name: '登录工作台' })
   fireEvent.change(screen.getByLabelText('用户名'), { target: { value: 'alice' } })
-  fireEvent.change(screen.getByLabelText('密码'), { target: { value: 'long-test-password' } })
+  fireEvent.change(screen.getByLabelText('密码'), { target: { value: '123456' } })
   fireEvent.click(screen.getByRole('button', { name: '登录' }))
   await screen.findByText('private workspace')
   fireEvent.click(screen.getByRole('button', { name: '退出登录' }))
@@ -66,6 +66,32 @@ it('requires an invitation and returns to login after registration', async () =>
   expect(screen.queryByText('private')).not.toBeInTheDocument()
 })
 
+it('uses a browser-valid username pattern without changing allowed names', async () => {
+  vi.stubGlobal('fetch', vi.fn().mockResolvedValue(reply({}, 401)))
+  render(<AccountGate><div>private</div></AccountGate>)
+  const input = await screen.findByLabelText('用户名')
+  const pattern = new RegExp(`^(?:${input.getAttribute('pattern')})$`, 'v')
+  for (const value of ['alice', 'qa_alice_123', 'test-user', 'A12']) expect(pattern.test(value)).toBe(true)
+  for (const value of ['ab', '-alice', 'with space', 'user@example.com', 'a'.repeat(33)]) expect(pattern.test(value)).toBe(false)
+})
+
+it('uses a six-character minimum for registration and both new-password fields', async () => {
+  const fetch = vi.fn().mockResolvedValue(reply({}, 401))
+  vi.stubGlobal('fetch', fetch)
+  const view = render(<AccountGate><div>private</div></AccountGate>)
+  fireEvent.click(await screen.findByRole('tab', { name: '注册' }))
+  const password = screen.getByLabelText('密码')
+  expect(password).toHaveAttribute('minlength', '6')
+  fireEvent.change(password, { target: { value: '123456' } })
+  expect((password as HTMLInputElement).checkValidity()).toBe(true)
+  view.unmount()
+  fetch.mockResolvedValue(reply({ enabled: true, user }))
+  render(<AccountGate><div>private</div></AccountGate>)
+  fireEvent.click(await screen.findByRole('button', { name: '修改密码' }))
+  expect(screen.getByLabelText('新密码')).toHaveAttribute('minlength', '6')
+  expect(screen.getByLabelText('确认新密码')).toHaveAttribute('minlength', '6')
+})
+
 it('fails closed when the identity service is unavailable', async () => {
   vi.stubGlobal('fetch', vi.fn().mockResolvedValue(reply({}, 503)))
   render(

packages/dsh-triton-riscv/frontend/src/AccountGate.tsx

@@ -220,7 +220,7 @@ export function AccountGate({ children }: { children: ReactNode }) {
                     required
                     minLength={3}
                     maxLength={32}
-                    pattern="[a-zA-Z0-9][a-zA-Z0-9_-]{2,31}"
+                    pattern="[a-zA-Z0-9][a-zA-Z0-9_\-]{2,31}"
                   />
                 </label>
                 <label>
@@ -230,7 +230,7 @@ export function AccountGate({ children }: { children: ReactNode }) {
                     name="password"
                     autoComplete={register ? 'new-password' : 'current-password'}
                     required
-                    minLength={register ? 12 : 1}
+                    minLength={register ? 6 : 1}
                     maxLength={128}
                   />
                 </label>
@@ -307,15 +307,15 @@ export function AccountGate({ children }: { children: ReactNode }) {
               </label>
               <label>
                 新密码
-                <input name="new" type="password" required minLength={12} maxLength={128} autoComplete="new-password" />
+                <input name="new" type="password" required minLength={6} maxLength={128} autoComplete="new-password" />
               </label>
               <label>
                 确认新密码
                 <input
                   name="confirm"
                   type="password"
                   required
-                  minLength={12}
+                  minLength={6}
                   maxLength={128}
                   autoComplete="new-password"
                 />