feat(core): implement persistent platform and supervised Harness

CHANGELOG.md

@@ -6,13 +6,15 @@ Delivered changes are recorded here. Version tags use `v` followed by a semantic
 
 ### Added
 
+- Initial TypeScript platform archive, generic single-agent Harness, supervised IPC worker and streaming model adapter, with confirmation, interaction validation, cancellation, restart reconciliation and immutable report archives.
+- Native Windows core tests, strict type checking and an explicit opt-in DeepSeek streaming/tool-call smoke test. Application UI, retrieval and full research acceptance remain pending.
 - Research workspace specification covering conversation, execution state, live artifact revisions, evidence navigation, responsive layouts and application acceptance scenarios.
 - Minimal OpenLab Harness specification and ADR 0002 selecting a self-developed first runtime, AG-UI at the Platform gateway and assistant-ui for conversation.
 
 ### Changed
 
 - Product, architecture, logging, architecture diagram and bilingual README now describe Platform-owned conversations, steps, human interactions and artifacts independently of runtime, model, tools and UI.
-- OpenCode becomes an optional future adapter; ADR 0001 preserves the historical selection with a supersession notice. This revision delivers documentation only, with no application code or new runtime dependency.
+- OpenCode becomes an optional future adapter; ADR 0001 preserves the historical selection with a supersession notice.
 
 ## [0.0.1] - 2026-10-24
 

README.md

@@ -19,6 +19,8 @@ OpenLab is a local-first research workspace being designed to connect open resea
 
 **v0.0.1 establishes repository governance and design documents. There is no runnable research application or installer in this release.** The commands below validate this repository; they do not launch OpenLab. Screenshots and research demonstrations will be added when the corresponding functionality is available.
 
+Current development source adds the platform archive, generic Harness and supervised worker. See [development](docs/development.md) and the [implementation plan](docs/implementation-plan.md) for implemented scope and remaining application gates.
+
 | Capability | Status |
 | --- | --- |
 | Contribution, issue, review, and release processes | Available in this repository |

README.zh-CN.md

@@ -1,5 +1,7 @@
 # OpenLab
 
+当前开发源码已加入平台档案、通用 Harness 和受监督的 worker。已实现范围、验证结果与剩余应用验收项见[开发说明](docs/development.md)和[实施计划](docs/implementation-plan.md)。v0.0.1 发布范围仍为仓库基础设施。
+
 <picture>
   <source media="(prefers-color-scheme: dark)" srcset="assets/brand/openlab-mark-inverse.svg">
   <img src="assets/brand/openlab-mark.svg" alt="OpenLab 烧瓶与波浪标识" width="112" height="112">

THIRD_PARTY_NOTICES.md

@@ -15,6 +15,8 @@ OpenLab's Apache-2.0 license applies to OpenLab-authored material. Referenced or
 
 The npm lockfile records exact development dependencies and integrity hashes. Their license metadata is available in the installed packages. CI downloads actionlint (MIT) and Gitleaks (MIT) from their official releases; pinned versions, archive URLs, and SHA-256 digests are recorded in [the tool manifest](scripts/tools.json). These executables are cached locally and are not included in the source release.
 
+The initial application core uses Node.js 24's built-in APIs, including SQLite and child processes, without an additional runtime npm dependency. Development adds TypeScript 7.0.2 (Apache-2.0) and `@types/node` 24.19.2 (MIT); exact transitive packages and integrity hashes are in the lockfile. Their package license metadata was inspected on 2026-10-24. The DeepSeek API is an external user-configured service; no DeepSeek implementation is bundled.
+
 GitHub Actions used by the workflows retain their upstream licenses and are pinned to immutable commit SHAs. Any future application distribution must include an updated dependency and license inventory.
 
 ## Planned application components and design references

app/contracts.ts

@@ -0,0 +1,56 @@
+export type Json = null | boolean | number | string | Json[] | { [key: string]: Json };
+export type ObjectData = { [key: string]: Json };
+export type RunState = 'queued' | 'running' | 'awaiting_confirmation' | 'awaiting_input' | 'awaiting_permission' | 'cancelling' | 'succeeded' | 'failed' | 'cancelled' | 'interrupted';
+export const terminal = new Set<RunState>(['succeeded', 'failed', 'cancelled', 'interrupted']);
+export interface ResearchEvent { runId: string; seq: number; at: string; kind: string; data: ObjectData }
+export interface Limits { turns: number; durationMs: number; callMs: number; tokens: number }
+export interface Run {
+  id: string; projectId: string; conversationId: string; requestKey: string; question: string;
+  state: RunState; phase: string; createdAt: string; config: ObjectData; limits: Limits;
+  plan?: { version: number; text: string; tools: string[]; confirmed: boolean };
+  interaction?: Interaction; decisions: Record<string, Decision>; parentRunId?: string;
+  error?: string;
+}
+export interface Interaction {
+  id: string; version: number; kind: 'confirmation' | 'question' | 'permission'; prompt: string;
+  input: 'text' | 'single_choice' | 'multiple_choice' | 'boolean'; options?: string[]; scope?: ObjectData;
+}
+export interface Decision { interactionId: string; version: number; value: Json }
+export interface ToolCall { id: string; name: string; arguments: string }
+export interface Message { role: 'system' | 'user' | 'assistant' | 'tool'; content: string; tool_calls?: { id: string; type: 'function'; function: { name: string; arguments: string } }[]; tool_call_id?: string }
+export interface ToolDefinition { name: string; description: string; parameters: ObjectData }
+export type ModelChunk = { type: 'text'; text: string } | { type: 'tool'; call: ToolCall } | { type: 'usage'; input: number; output: number };
+export interface ModelProvider {
+  stream(messages: Message[], tools: ToolDefinition[], signal: AbortSignal): AsyncIterable<ModelChunk>;
+}
+export interface ToolExecutor {
+  definitions: ToolDefinition[];
+  execute(call: ToolCall, signal: AbortSignal): Promise<Json>;
+}
+export interface HarnessHost {
+  emit(kind: string, data: ObjectData): Promise<void>;
+  wait(interaction: Omit<Interaction, 'id' | 'version'>, signal: AbortSignal): Promise<Json>;
+}
+export interface RuntimeCapabilities { streaming: boolean; cancellation: boolean; questions: boolean; permissions: boolean; checkpoints: false }
+export interface RuntimeAdapter {
+  capabilities(): RuntimeCapabilities;
+  start(run: Run, host: HarnessHost): Promise<void>;
+  cancel(runId: string): Promise<void>;
+  close(): Promise<void>;
+}
+export class DomainError extends Error {
+  status: number;
+  constructor(message: string, status = 400) { super(message); this.name = 'DomainError'; this.status = status; }
+}
+export function object(value: unknown): ObjectData {
+  if (!value || typeof value !== 'object' || Array.isArray(value)) throw new DomainError('Expected an object');
+  return value as ObjectData;
+}
+export function text(value: unknown, name = 'text', max = 100_000): string {
+  if (typeof value !== 'string' || !value.trim() || value.length > max) throw new DomainError(`Invalid ${name}`);
+  return value;
+}
+export function id(value: string): string {
+  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(value)) throw new DomainError('Invalid identifier');
+  return value;
+}

app/harness.ts

@@ -0,0 +1,82 @@
+import { randomUUID } from 'node:crypto';
+import { DomainError } from './contracts.ts';
+import type { HarnessHost, Limits, Message, ModelProvider, ToolExecutor } from './contracts.ts';
+
+export interface ContextStrategy { assemble(messages: Message[]): { messages: Message[]; trimmed: number } }
+export const fullContext: ContextStrategy = { assemble: messages => ({ messages, trimmed: 0 }) };
+
+// This loop knows nothing about SQLite, HTTP, scholarly APIs or the browser.
+export async function executeHarness(options: {
+  question: string; instructions: string; model: ModelProvider; tools: ToolExecutor;
+  host: HarnessHost; limits: Limits; signal: AbortSignal; context?: ContextStrategy;
+}): Promise<void> {
+  const { host, model, tools, limits } = options;
+  const signal = AbortSignal.any([options.signal, AbortSignal.timeout(limits.durationMs)]);
+  let turns = 0;
+  let used = 0;
+  let unknownUsage = false;
+  const messages: Message[] = [
+    { role: 'system', content: options.instructions },
+    { role: 'user', content: options.question },
+  ];
+  const call = async (planning: boolean): Promise<Message> => {
+    signal.throwIfAborted();
+    if (turns >= limits.turns || used >= limits.tokens) throw new DomainError('Run usage/turn limit reached');
+    turns++;
+    const assembled = (options.context ?? fullContext).assemble(messages);
+    if (assembled.trimmed) await host.emit('context_trimmed', { count: assembled.trimmed });
+    const messageId = randomUUID();
+    const stepId = randomUUID();
+    await host.emit('step_started', { stepId, label: planning ? 'Planning' : 'Research', turn: turns });
+    await host.emit('message_start', { messageId, role: 'assistant', stepId });
+    const reply: Message = { role: 'assistant', content: '' };
+    let usageReported = false;
+    for await (const chunk of model.stream(assembled.messages, planning ? [] : tools.definitions, AbortSignal.any([signal, AbortSignal.timeout(limits.callMs)]))) {
+      signal.throwIfAborted();
+      if (chunk.type === 'text') {
+        reply.content += chunk.text;
+        await host.emit('message_delta', { messageId, text: chunk.text, stepId });
+      } else if (chunk.type === 'tool') {
+        if (planning) throw new DomainError('Model requested tools during planning');
+        if (!chunk.call.id || reply.tool_calls?.some(c => c.id === chunk.call.id)) throw new DomainError('Invalid or duplicate tool call identifier');
+        (reply.tool_calls ??= []).push({ id: chunk.call.id, type: 'function', function: { name: chunk.call.name, arguments: chunk.call.arguments } });
+      } else {
+        if (!Number.isSafeInteger(chunk.input) || !Number.isSafeInteger(chunk.output) || chunk.input < 0 || chunk.output < 0 || usageReported) throw new DomainError('Invalid model usage');
+        usageReported = true;
+        used += chunk.input + chunk.output;
+        await host.emit('usage', { input: chunk.input, output: chunk.output, totalReported: used, unknownUsage });
+      }
+    }
+    if (!usageReported) {
+      unknownUsage = true;
+      await host.emit('usage', { input: null, output: null, totalReported: used, unknownUsage: true });
+    }
+    if (!reply.content.trim() && !reply.tool_calls?.length) throw new DomainError('Empty model response');
+    await host.emit('message_end', { messageId, content: reply.content, stepId });
+    await host.emit('step_finished', { stepId });
+    messages.push(reply);
+    return reply;
+  };
+  messages.push({ role: 'user', content: 'First propose a concise research brief and plan in the user language. State evidence limitations. Do not research yet. Research tools require user confirmation.' });
+  const plan = await call(true);
+  const confirmed = await host.wait({ kind: 'confirmation', input: 'boolean', prompt: plan.content, scope: { tools: tools.definitions.map(t => t.name) } }, signal);
+  if (confirmed !== true) throw new DomainError('Research plan was not confirmed');
+  messages.push({ role: 'user', content: 'The displayed plan and listed tools are confirmed. Execute it within scope. Save a report with evidence and limitations using the report tool before finishing.' });
+  while (true) {
+    const reply = await call(false);
+    if (!reply.tool_calls?.length) return;
+    for (const tool of reply.tool_calls) {
+      signal.throwIfAborted();
+      const call = { id: tool.id, name: tool.function.name, arguments: tool.function.arguments };
+      if (!tools.definitions.some(t => t.name === call.name)) throw new DomainError('Model requested an unregistered tool');
+      const stepId = randomUUID();
+      await host.emit('tool_intent', { stepId, callId: call.id, name: call.name, arguments: call.arguments });
+      // Only the tool host can validate and authorize arguments. Retrieved text
+      // and model output cannot expand the tool registry or its capabilities.
+      const result = await tools.execute(call, signal);
+      signal.throwIfAborted();
+      await host.emit('tool_result', { stepId, callId: call.id, result });
+      messages.push({ role: 'tool', tool_call_id: call.id, content: JSON.stringify(result) });
+    }
+  }
+}

app/model.ts

@@ -0,0 +1,105 @@
+import { DomainError, object, text } from './contracts.ts';
+import type { Message, ModelChunk, ModelProvider, ObjectData, ToolCall, ToolDefinition } from './contracts.ts';
+
+export interface ProviderConfig { baseUrl: string; model: string; apiKey: string; maxOutputTokens?: number; extra?: ObjectData }
+// SSE lines and UTF-8 characters may cross network chunks. A completed stream
+// needs [DONE] and a finish reason, not just a closed socket.
+export async function* sseData(response: Response): AsyncIterable<string> {
+  if (!response.body) throw new DomainError('Provider response has no body');
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = '';
+  let lines: string[] = [];
+  try {
+    while (true) {
+      const part = await reader.read();
+      buffer += decoder.decode(part.value, { stream: !part.done });
+      let end: number;
+      while ((end = buffer.indexOf('\n')) !== -1) {
+        const line = buffer.slice(0, end).replace(/\r$/, '');
+        buffer = buffer.slice(end + 1);
+        if (!line) { if (lines.length) yield lines.join('\n'); lines = []; }
+        else if (line.startsWith('data:')) lines.push(line.slice(5).replace(/^ /, ''));
+      }
+      if (buffer.length > 2_000_000) throw new DomainError('Provider event too large');
+      if (part.done) break;
+    }
+    if (buffer || lines.length) throw new DomainError('Truncated provider event stream');
+  } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); }
+}
+
+export class CompatibleModel implements ModelProvider {
+  config: ProviderConfig;
+  fetcher: typeof fetch;
+  constructor(config: ProviderConfig, fetcher = fetch) {
+    const url = new URL(config.baseUrl);
+    if (url.protocol !== 'https:' && !(url.protocol === 'http:' && ['127.0.0.1', 'localhost', '[::1]'].includes(url.hostname))) throw new DomainError('Model endpoint must use HTTPS or explicit loopback HTTP');
+    if (url.username || url.password || url.search || url.hash) throw new DomainError('Invalid model endpoint');
+    this.config = config;
+    this.fetcher = fetcher;
+  }
+  async *stream(messages: Message[], tools: ToolDefinition[], signal: AbortSignal): AsyncIterable<ModelChunk> {
+    const response = await this.fetcher(`${this.config.baseUrl.replace(/\/$/, '')}/chat/completions`, {
+      method: 'POST', signal, redirect: 'error',
+      headers: { 'content-type': 'application/json', authorization: `Bearer ${this.config.apiKey}` },
+      body: JSON.stringify({ ...this.config.extra, model: this.config.model, messages, stream: true, stream_options: { include_usage: true }, max_tokens: this.config.maxOutputTokens ?? 4096,
+        ...(tools.length ? { tools: tools.map(tool => ({ type: 'function', function: tool })) } : {}),
+      }),
+    });
+    // Provider bodies may echo the Authorization header. Never archive raw errors.
+    if (!response.ok) {
+      await response.body?.cancel();
+      const retry = response.headers.get('retry-after');
+      throw new DomainError(`Model request failed (HTTP ${response.status}${retry ? '; retry guidance available' : ''})`, 502);
+    }
+    if (!response.headers.get('content-type')?.includes('text/event-stream')) throw new DomainError('Model endpoint did not return SSE');
+    const calls = new Map<number, ToolCall>();
+    let finished = false;
+    let done = false;
+    let bytes = 0;
+    for await (const payload of sseData(response)) {
+      signal.throwIfAborted();
+      if (payload === '[DONE]') { done = true; break; }
+      bytes += payload.length;
+      if (bytes > 4_000_000) throw new DomainError('Model output limit exceeded');
+      const chunk = object(JSON.parse(payload));
+      if (chunk.error) throw new DomainError('Provider reported a stream error');
+      if (chunk.usage) {
+        const usage = object(chunk.usage);
+        yield { type: 'usage', input: Number(usage.prompt_tokens), output: Number(usage.completion_tokens) };
+      }
+      if (!Array.isArray(chunk.choices)) throw new DomainError('Malformed model chunk');
+      for (const raw of chunk.choices) {
+        const choice = object(raw);
+        if (choice.index !== 0) throw new DomainError('Unexpected model choice');
+        const delta = object(choice.delta ?? {});
+        if (delta.content !== undefined && delta.content !== null) {
+          if (typeof delta.content !== 'string') throw new DomainError('Invalid model text');
+          if (delta.content) yield { type: 'text', text: delta.content };
+        }
+        if (Array.isArray(delta.tool_calls)) for (const value of delta.tool_calls) {
+          const fragment = object(value);
+          const index = Number(fragment.index);
+          if (!Number.isSafeInteger(index) || index < 0 || index > 31) throw new DomainError('Invalid tool index');
+          const call = calls.get(index) ?? { id: '', name: '', arguments: '' };
+          if (fragment.id) call.id += text(fragment.id);
+          if (fragment.function) {
+            const fn = object(fragment.function);
+            if (typeof fn.name === 'string') call.name += fn.name;
+            if (typeof fn.arguments === 'string') call.arguments += fn.arguments;
+          }
+          calls.set(index, call);
+        }
+        if (choice.finish_reason) {
+          if (!['stop', 'tool_calls'].includes(String(choice.finish_reason))) throw new DomainError(`Model stopped before completion: ${String(choice.finish_reason)}`);
+          finished = true;
+        }
+      }
+    }
+    if (!done || !finished) throw new DomainError('Model stream ended without confirmed completion');
+    for (const call of calls.values()) {
+      text(call.id); text(call.name); object(JSON.parse(call.arguments));
+      yield { type: 'tool', call };
+    }
+  }
+}