diff --git a/packages/dsh-multi-agent/README.md b/packages/dsh-multi-agent/README.md index e5c5f33..76b15ec 100644 --- a/packages/dsh-multi-agent/README.md +++ b/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 diff --git a/packages/dsh-multi-agent/python/dsh_multi_agent/__init__.py b/packages/dsh-multi-agent/python/dsh_multi_agent/__init__.py index a31bddd..a794fd2 100644 --- a/packages/dsh-multi-agent/python/dsh_multi_agent/__init__.py +++ b/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"] diff --git a/packages/dsh-multi-agent/python/dsh_multi_agent/parallel.py b/packages/dsh-multi-agent/python/dsh_multi_agent/parallel.py new file mode 100644 index 0000000..ba8b130 --- /dev/null +++ b/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] diff --git a/packages/dsh-multi-agent/python/dsh_multi_agent/routing.py b/packages/dsh-multi-agent/python/dsh_multi_agent/routing.py new file mode 100644 index 0000000..6bad50e --- /dev/null +++ b/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) diff --git a/packages/dsh-multi-agent/tests/unit/parallel.spec.ts b/packages/dsh-multi-agent/tests/unit/parallel.spec.ts new file mode 100644 index 0000000..33c8782 --- /dev/null +++ b/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) +}) diff --git a/packages/dsh-multi-agent/tests/unit/routing.spec.ts b/packages/dsh-multi-agent/tests/unit/routing.spec.ts new file mode 100644 index 0000000..35d4299 --- /dev/null +++ b/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 }) +}) diff --git a/packages/dsh-triton-riscv/.gitignore b/packages/dsh-triton-riscv/.gitignore index 1148169..585207c 100644 --- a/packages/dsh-triton-riscv/.gitignore +++ b/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 diff --git a/packages/dsh-triton-riscv/README.md b/packages/dsh-triton-riscv/README.md index 663a375..7f48855 100644 --- a/packages/dsh-triton-riscv/README.md +++ b/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 diff --git a/packages/dsh-triton-riscv/frontend/src/AccountGate.test.tsx b/packages/dsh-triton-riscv/frontend/src/AccountGate.test.tsx index 8c14cf9..d1ef7b8 100644 --- a/packages/dsh-triton-riscv/frontend/src/AccountGate.test.tsx +++ b/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(
private
) + 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(
private
) + 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(
private
) + 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( diff --git a/packages/dsh-triton-riscv/frontend/src/AccountGate.tsx b/packages/dsh-triton-riscv/frontend/src/AccountGate.tsx index 7e0559b..a0f4f3a 100644 --- a/packages/dsh-triton-riscv/frontend/src/AccountGate.tsx +++ b/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}" /> @@ -307,7 +307,7 @@ export function AccountGate({ children }: { children: ReactNode }) {