From 760d102d6044fccdcf7c598a3cb1c2f70374f009 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 02:19:32 -0800 Subject: [PATCH 1/7] feat(evals): evaluate independent thread suggestions --- evals/README.md | 1 + evals/thread-split/README.md | 53 +++++++++++++++++++++++++ evals/thread-split/detector.py | 53 +++++++++++++++++++++++++ evals/thread-split/requirements.txt | 1 + evals/thread-split/run.py | 61 +++++++++++++++++++++++++++++ evals/thread-split/test_detector.py | 32 +++++++++++++++ 6 files changed, 201 insertions(+) create mode 100644 evals/thread-split/README.md create mode 100644 evals/thread-split/detector.py create mode 100644 evals/thread-split/requirements.txt create mode 100644 evals/thread-split/run.py create mode 100644 evals/thread-split/test_detector.py diff --git a/evals/README.md b/evals/README.md index 1aa5796b00378..1eafbe0932e8a 100644 --- a/evals/README.md +++ b/evals/README.md @@ -8,6 +8,7 @@ Write or run an eval when a change affects what a model sees or does: agent inst | ------------------------------- | ---------------------------------------------------------------------------------------------------- | | [mcp-tool-use](mcp-tool-use/) | Whether agents use T3's MCP tools correctly, including device routing and recovery, in fake threads. | | [thread-titles](thread-titles/) | Whether generated thread titles name the right subject. | +| [thread-split](thread-split/) | Offline local-model evaluation of whether a message can start an independent thread. | | [test-signal](test-signal/) | Whether a cheap model running the test-signal check flags low-signal tests. | ## Run an eval diff --git a/evals/thread-split/README.md b/evals/thread-split/README.md new file mode 100644 index 0000000000000..e393ae58bd06e --- /dev/null +++ b/evals/thread-split/README.md @@ -0,0 +1,53 @@ +# Thread independence evaluation + +Tests whether a local decision model can identify a new message that needs none +of the preceding thread's results, files, decisions or instructions. A different +subject alone does not establish independence. This is an offline experiment; +it does not create threads, alter history, or display suggestions. + +The direct prompt lives in `detector.py`. Each request contains one question: +pplx's joint question attention makes scores depend on siblings when batched. +Truncated requests and inference failures abstain and are counted separately. + +## Run + +Use Python 3.13 on Apple Silicon with `mlx-decision==1.1.0` and a local model. +This evaluation uses Python because the measured runtime is MLX; it does not +call a paid provider or download model weights. + +```sh +python3 -m unittest discover -s evals/thread-split +python3 evals/thread-split/run.py --cases /path/to/private-cases.json \ + --model /path/to/pplx-decider-flash-v1.2 --threshold 0.94 \ + --out /path/to/new-run +``` + +The output directory must not exist. It receives scores, abstentions, request +latencies, the exact prompt, input digest, threshold and confusion counts. +The threshold is explicit; this runner never tunes it on evaluation labels. + +Cases are a JSON array with `id`, `session`, `label` (`INDEPENDENT` or `DEPENDS`), +and `state` containing only `prior_thread` and `new_message`. Prior thread entries +contain `user` and `assistant_final`. The new message's answer must never enter +state. Keep private transcripts outside the repository. Use only synthetic, +anonymized cases in committed fixtures. + +Choose thresholds using separate calibration sessions; evaluate on wholly new +sessions. Label before scoring, keep ambiguous labels visible, and report real +messages separately from synthetic task switches. Repeat model samples before +drawing conclusions. Abstentions are neither correct predictions nor negatives. + +## Evidence and draft boundary + +The exploratory November 3 pilot had 190 messages from 44 sessions, with only +nine independent messages. Pplx/direct caught three of nine with one false flag +among 181 dependent messages at the sample-selected cutoff 0.938570. Median +latency was 0.898 seconds; p95 was 5.614 seconds. Those counts were calibrated +and measured on the same sample, so they do not establish held-out accuracy. +A separate synthetic-switch analysis used a different cutoff (about 0.919); +do not combine those thresholds or treat either as a production default. + +Before app integration, validate on new sessions and decide how the local model +is provisioned on each server platform. Any initial client affordance should be +opt-in, passive and asynchronous, with explicit user action to create a fresh +thread. Automatic splitting and interrupting prompts are outside this draft. diff --git a/evals/thread-split/detector.py b/evals/thread-split/detector.py new file mode 100644 index 0000000000000..1d8d1377eb373 --- /dev/null +++ b/evals/thread-split/detector.py @@ -0,0 +1,53 @@ +import math + +PROMPT = ( + 'Could the agent handle the new user message independently of all earlier thread ' + 'context, without reusing any earlier result, file, decision, or instruction? ' + 'Answer yes only if independent.' +) + + +def request(state): + return { + 'state': state, + 'questions': {'q': { + 'type': 'noul', + 'instructions': PROMPT, + 'criteria': { + 'true': 'No earlier thread result, file, decision, or instruction is needed.', + 'false': 'Handling relies on an earlier thread result, file, decision, or instruction.', + }, + }}, + } + + +def score_response(response): + if response.truncated: + return None + score = float(response.answers['q'].noul) + if not math.isfinite(score) or not 0 <= score <= 1: + raise ValueError('Invalid independence score') + return score + + +def summarize(rows, threshold): + if not math.isfinite(threshold) or not 0 <= threshold <= 1: + raise ValueError('Threshold must be between zero and one') + counts = dict(tp=0, fp=0, tn=0, fn=0, abstained=0) + seen = set() + for row in rows: + if row['id'] in seen: + raise ValueError('Duplicate case id') + seen.add(row['id']) + if row['label'] not in ('INDEPENDENT', 'DEPENDS'): + raise ValueError('Unknown label') + score = row['score'] + if score is None: + counts['abstained'] += 1 + continue + if not math.isfinite(score) or not 0 <= score <= 1: + raise ValueError('Invalid independence score') + positive = row['label'] == 'INDEPENDENT' + predicted = score >= threshold + counts[('t' if positive == predicted else 'f') + ('p' if predicted else 'n')] += 1 + return counts diff --git a/evals/thread-split/requirements.txt b/evals/thread-split/requirements.txt new file mode 100644 index 0000000000000..e45b8c867d749 --- /dev/null +++ b/evals/thread-split/requirements.txt @@ -0,0 +1 @@ +mlx-decision==1.1.0 diff --git a/evals/thread-split/run.py b/evals/thread-split/run.py new file mode 100644 index 0000000000000..1ede92d9b5d8a --- /dev/null +++ b/evals/thread-split/run.py @@ -0,0 +1,61 @@ +import argparse +import hashlib +import json +from pathlib import Path +import time + +from detector import PROMPT, request, score_response, summarize + + +def main(): + parser = argparse.ArgumentParser(description='Offline thread-independence evaluation') + parser.add_argument('--cases', type=Path, required=True) + parser.add_argument('--model', type=Path, required=True) + parser.add_argument('--threshold', type=float, required=True) + parser.add_argument('--out', type=Path, required=True) + args = parser.parse_args() + summarize([], args.threshold) + cases = json.loads(args.cases.read_text()) + summarize([dict(id=c['id'], label=c['label'], score=None) for c in cases], args.threshold) + for case in cases: + state = case['state'] + if set(state) != {'prior_thread', 'new_message'}: + raise ValueError('State must contain only prior_thread and new_message') + if not isinstance(state['new_message'], str) or not isinstance(state['prior_thread'], list): + raise ValueError('Invalid thread state') + for turn in state['prior_thread']: + if set(turn) != {'user', 'assistant_final'} or not all(isinstance(v, str) for v in turn.values()): + raise ValueError('Prior turns must contain only user and assistant_final text') + args.out.mkdir(parents=True, exist_ok=False) + import mlx_decision + + model = mlx_decision.load(str(args.model)) + rows = [] + with (args.out / 'scores.jsonl').open('x') as output: + for case in cases: + state = case['state'] + row = {key: case[key] for key in ('id', 'session', 'label')} + start = time.perf_counter() + try: + row['score'] = score_response(model.decide_request(request(state))) + except Exception as error: + row.update(score=None, error=type(error).__name__) + row['seconds'] = time.perf_counter() - start + rows.append(row) + output.write(json.dumps(row) + '\n') + output.flush() + report = { + 'model': args.model.name, + 'threshold': args.threshold, + 'prompt': PROMPT, + 'cases_sha256': hashlib.sha256(args.cases.read_bytes()).hexdigest(), + 'counts': summarize(rows, args.threshold), + 'sessions': len({row['session'] for row in rows}), + 'scope': 'Offline evaluation only; no thread mutation or runtime integration', + } + (args.out / 'summary.json').write_text(json.dumps(report, indent=2) + '\n') + print(json.dumps(report)) + + +if __name__ == '__main__': + main() diff --git a/evals/thread-split/test_detector.py b/evals/thread-split/test_detector.py new file mode 100644 index 0000000000000..52d90d7dabba9 --- /dev/null +++ b/evals/thread-split/test_detector.py @@ -0,0 +1,32 @@ +import unittest +from types import SimpleNamespace + +from detector import score_response, summarize + + +class DetectorTest(unittest.TestCase): + def test_truncation_abstains_even_when_score_would_suggest_split(self): + response = SimpleNamespace(truncated=True, answers={'q': SimpleNamespace(noul=1.0)}) + self.assertIsNone(score_response(response)) + for invalid in (float('nan'), float('inf'), -0.1, 1.1): + response = SimpleNamespace(truncated=False, answers={'q': SimpleNamespace(noul=invalid)}) + with self.assertRaises(ValueError): + score_response(response) + + def test_confusion_counts_include_boundary_and_report_abstentions(self): + rows = [ + dict(id='switch', label='INDEPENDENT', score=0.94), + dict(id='followup', label='DEPENDS', score=0.95), + dict(id='miss', label='INDEPENDENT', score=0.2), + dict(id='keep', label='DEPENDS', score=0.2), + dict(id='error', label='INDEPENDENT', score=None), + ] + self.assertEqual(summarize(rows, 0.94), dict(tp=1, fp=1, tn=1, fn=1, abstained=1)) + with self.assertRaises(ValueError): + summarize(rows + rows[:1], 0.94) + with self.assertRaises(ValueError): + summarize([dict(id='bad', label='UNKNOWN', score=0.8)], 0.94) + + +if __name__ == '__main__': + unittest.main() From 6d3bc416afbfa4ebf1f8114151d6318e19efc17b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 09:30:26 -0800 Subject: [PATCH 2/7] feat(chat): suggest fresh threads for independent drafts --- .../settings/ThreadSettingsSections.tsx | 45 +++++ .../src/features/threads/ThreadComposer.tsx | 28 ++- .../threads/ThreadSplitSuggestion.tsx | 96 ++++++++++ .../src/observability/RpcInstrumentation.ts | 1 + .../ThreadSplitSuggestions.test.ts | 181 ++++++++++++++++++ .../ThreadSplitSuggestions.ts | 151 +++++++++++++++ apps/server/src/server.ts | 2 + apps/server/src/ws.test.ts | 19 +- apps/server/src/ws.ts | 4 + apps/web/src/components/chat/ChatComposer.tsx | 17 ++ .../ThreadSplitSuggestion.browser.test.tsx | 137 +++++++++++++ .../components/chat/ThreadSplitSuggestion.tsx | 133 +++++++++++++ .../components/settings/LabsSettingsPanel.tsx | 46 +++++ .../src/components/settings/settingsSearch.ts | 7 + docs/user/composer.md | 26 +++ evals/thread-split/README.md | 11 +- evals/thread-split/detector.py | 22 +-- .../src/state/commandPermissions.test.ts | 10 + .../src/state/threadCommands.ts | 4 + .../contracts/src/clientRpcPermissions.ts | 1 + packages/contracts/src/index.ts | 2 + packages/contracts/src/rpc.ts | 9 + packages/contracts/src/settings.ts | 6 + packages/contracts/src/threadSplit.ts | 9 + packages/shared/package.json | 1 + packages/shared/src/threadSplitPrompt.json | 8 + 26 files changed, 953 insertions(+), 23 deletions(-) create mode 100644 apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx create mode 100644 apps/server/src/orchestration-v2/ThreadSplitSuggestions.test.ts create mode 100644 apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts create mode 100644 apps/web/src/components/chat/ThreadSplitSuggestion.browser.test.tsx create mode 100644 apps/web/src/components/chat/ThreadSplitSuggestion.tsx create mode 100644 packages/contracts/src/threadSplit.ts create mode 100644 packages/shared/src/threadSplitPrompt.json diff --git a/apps/mobile/src/features/settings/ThreadSettingsSections.tsx b/apps/mobile/src/features/settings/ThreadSettingsSections.tsx index 4952f9a2f20ca..ab76ad7483095 100644 --- a/apps/mobile/src/features/settings/ThreadSettingsSections.tsx +++ b/apps/mobile/src/features/settings/ThreadSettingsSections.tsx @@ -42,6 +42,7 @@ export function SettingsLabsRouteScreen() { contentContainerClassName="gap-6 px-5 pt-4" contentContainerStyle={{ paddingBottom: Math.max(insets.bottom, 18) + 18 }} > + @@ -294,3 +295,47 @@ function LegacySettingsSection() { ); } + +function ThreadSplitSettingsSection() { + const { environments } = useEnvironments(); + const connected = environments.filter( + (environment) => environment.connection.phase === "connected", + ); + const writable = useEnvironmentsWithScope(connected, AuthSettingsWriteScope); + const updateSettings = useAtomCommand(serverEnvironment.updateSettings, { + label: "thread split suggestions setting", + reportFailure: true, + }); + return ( + + {connected.map((environment) => { + const policy = environment.serverConfig?.managedPolicy; + const managed = isManagedSettingPath(policy, "threadSplitSuggestionsEnabled"); + return ( + + void updateSettings({ + environmentId: environment.environmentId, + input: { patch: { threadSplitSuggestionsEnabled: value } }, + }) + } + /> + ); + })} + + ); +} diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index b8540d45cfec0..949882f68c6e9 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -1,3 +1,4 @@ +import { ThreadSplitSuggestion } from "./ThreadSplitSuggestion"; import { ProjectKnowledgeEntry } from "../../components/ProjectKnowledgeEntry"; import { useSidebarWorkspaceTabs } from "../workspace/useSystemWorkspaceTabBar"; import { Button } from "../../components/ui/Button"; @@ -449,7 +450,11 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps const [previewVideo, setPreviewVideo] = useState(null); const composerOwnerKey = scopedThreadKey(props.environmentId, props.selectedThread.id); const composerDraftKey = props.draftKey ?? composerOwnerKey; - const { text: draftMessage, attachments: draftAttachments } = useComposerDraft(composerDraftKey); + const { + text: draftMessage, + attachments: draftAttachments, + context: draftContext, + } = useComposerDraft(composerDraftKey); const queuedEdit = props.queuedEdit ?? null; const hasContent = draftMessage.trim().length > 0 || @@ -903,6 +908,27 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps /> ) : null} + + + ); +} + +function ActiveThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { + const navigation = useNavigation(); + const [suggestedMessage, setSuggestedMessage] = useState(null); + const suggest = useAtomCommand(threadEnvironment.suggestSplit, { + reportFailure: false, + reportDefect: false, + }); + const request = useEffectEvent(suggest); + const { environmentId, threadId, message } = props; + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- debounce remote inference and discard responses after the draft or destination changes + useEffect(() => { + let cancelled = false; + const timer = setTimeout(() => { + void request({ environmentId, input: { threadId, message } }).then((result) => { + if (!cancelled) + setSuggestedMessage(result._tag === "Success" && result.value.suggest ? message : null); + }); + }, 1_000); + return () => { + cancelled = true; + clearTimeout(timer); + }; + }, [environmentId, threadId, message]); + + if (suggestedMessage !== message) return null; + + return ( + + + + + ); +} diff --git a/apps/web/src/components/settings/LabsSettingsPanel.tsx b/apps/web/src/components/settings/LabsSettingsPanel.tsx index 36c289b8335bc..cadd76b8b67ae 100644 --- a/apps/web/src/components/settings/LabsSettingsPanel.tsx +++ b/apps/web/src/components/settings/LabsSettingsPanel.tsx @@ -5,6 +5,7 @@ import { managedCapabilityDenial } from "@t3tools/contracts"; import { hasCloudPublicConfig } from "~/cloud/publicConfig"; import { useT3ConnectAuthPrompt } from "../clerk/useT3ConnectAuthPrompt"; import { Button } from "../ui/button"; +import { Input } from "../ui/input"; import { Switch } from "../ui/switch"; import { ScopedSwitch } from "./ScopedSwitch"; import { SettingsPageContainer, SettingsRow, SettingsSection } from "./settingsLayout"; @@ -19,6 +20,7 @@ export function LabsSettingsPanel() { return ( + {/* Clerk is only mounted when this build has T3 Connect. */} @@ -203,3 +205,47 @@ function SimlockHostsRow() { /> ); } + +function ThreadSplitSuggestionsRow() { + const settings = useScopedSettings(); + const updateSettings = useUpdateScopedSettings(); + return ( + <> + + updateSettings({ threadSplitSuggestionsEnabled: Boolean(checked) }) + } + aria-label="Suggest new threads" + /> + } + /> + { + const value = event.target.value.trim(); + if (value !== settings.threadSplitModelEndpoint) + updateSettings({ threadSplitModelEndpoint: value }); + }} + /> + } + /> + + ); +} diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 3ee0d1463c116..7e08e052a23aa 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -458,6 +458,13 @@ export const SETTINGS_SEARCH_ITEMS = [ to: "/settings/general", searchTerms: ["hide fold running monitoring threads inbox sidebar shelf"], }, + { + id: "thread-split-suggestions", + title: "Suggest new threads", + to: "/settings/labs", + scope: "environment", + searchTerms: ["split independent topic decision model"], + }, { id: "boring-sidebar", title: "Boring sidebar (group by project)", diff --git a/docs/user/composer.md b/docs/user/composer.md index 35a4b858ae80a..663352c273807 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -453,3 +453,29 @@ including filenames or folders containing `#` or `?`. On mobile, select a PDF attachment or link to open it. iOS uses the native viewer; Android opens a compatible installed file viewer. + +## Suggestions to start a fresh thread + +In **Settings → Labs**, enable **Thread split suggestions** for an environment +and configure its local decision-model endpoint. Suggestions are off by default. +When an idle thread has a text-only draft that appears independent of its earlier +work, T3 can offer to start a new thread with that draft. Nothing is sent until +you send it yourself. Dismiss the suggestion to keep working in the current thread, +or turn the setting off to stop checks. + +Checks run on the connected server, including when you use a remote browser or +phone. The experimental model can miss new tasks or suggest a split for a follow-up; +you decide whether the earlier context is still needed. Attachments and context +references stay in their current thread and are not eligible for this suggestion. + +The endpoint must be a loopback Jev-compatible endpoint on the server machine, +using `pplx-decider-flash-v1.2`. For an existing MLX Decision installation on Apple +Silicon, start the model with: + +```sh +mlx-decision server --model /path/to/pplx-decider-flash-v1.2 --host 127.0.0.1 --port 8000 +``` + +Enter `http://127.0.0.1:8000/v1/systemone` in the endpoint setting. +T3 does not install or download a model. When the endpoint is unavailable, checks +quietly abstain and the composer continues working normally. diff --git a/evals/thread-split/README.md b/evals/thread-split/README.md index e393ae58bd06e..4db5a82862c7a 100644 --- a/evals/thread-split/README.md +++ b/evals/thread-split/README.md @@ -5,7 +5,8 @@ of the preceding thread's results, files, decisions or instructions. A different subject alone does not establish independence. This is an offline experiment; it does not create threads, alter history, or display suggestions. -The direct prompt lives in `detector.py`. Each request contains one question: +The direct prompt is shared with the server in +`packages/shared/src/threadSplitPrompt.json`. Each request contains one question: pplx's joint question attention makes scores depend on siblings when batched. Truncated requests and inference failures abstain and are counted separately. @@ -47,7 +48,7 @@ and measured on the same sample, so they do not establish held-out accuracy. A separate synthetic-switch analysis used a different cutoff (about 0.919); do not combine those thresholds or treat either as a production default. -Before app integration, validate on new sessions and decide how the local model -is provisioned on each server platform. Any initial client affordance should be -opt-in, passive and asynchronous, with explicit user action to create a fresh -thread. Automatic splitting and interrupting prompts are outside this draft. +The app integration is opt-in, passive and asynchronous, using an explicitly +configured local endpoint. It requires user action to move a draft into a fresh +thread. Validate on new sessions before considering a default-on rollout. +Automatic splitting and interrupting prompts are outside this feature. diff --git a/evals/thread-split/detector.py b/evals/thread-split/detector.py index 1d8d1377eb373..faafe0049e52a 100644 --- a/evals/thread-split/detector.py +++ b/evals/thread-split/detector.py @@ -1,24 +1,16 @@ import math -PROMPT = ( - 'Could the agent handle the new user message independently of all earlier thread ' - 'context, without reusing any earlier result, file, decision, or instruction? ' - 'Answer yes only if independent.' +import json +from pathlib import Path + +QUESTION = json.loads( + (Path(__file__).resolve().parents[2] / 'packages/shared/src/threadSplitPrompt.json').read_text() ) +PROMPT = QUESTION['instructions'] def request(state): - return { - 'state': state, - 'questions': {'q': { - 'type': 'noul', - 'instructions': PROMPT, - 'criteria': { - 'true': 'No earlier thread result, file, decision, or instruction is needed.', - 'false': 'Handling relies on an earlier thread result, file, decision, or instruction.', - }, - }}, - } + return {'state': state, 'questions': {'q': QUESTION}} def score_response(response): diff --git a/packages/client-runtime/src/state/commandPermissions.test.ts b/packages/client-runtime/src/state/commandPermissions.test.ts index 55a40711aa46c..3f612ebe05f2f 100644 --- a/packages/client-runtime/src/state/commandPermissions.test.ts +++ b/packages/client-runtime/src/state/commandPermissions.test.ts @@ -72,11 +72,21 @@ describe("command permissions", () => { Effect.gen(function* () { const registry = yield* setup; const mutate = createCommandPermissions(runtime, WS_METHODS.projectKnowledgeMutate); + const suggestSplit = createCommandPermissions( + runtime, + WS_METHODS.orchestrationSuggestThreadSplit, + ); registry.set(sessions(env), AsyncResult.success(grant(true))); registry.set(sessions(other), AsyncResult.success(grant(false))); expect(registry.get(mutate.permissionAtom(env))).toBe(true); expect(registry.get(mutate.permissionAtom(other))).toBe(false); yield* mutate.authorize(registry, env); + expect(registry.get(suggestSplit.permissionAtom(env))).toBe(true); + expect(registry.get(suggestSplit.permissionAtom(other))).toBe(false); + yield* suggestSplit.authorize(registry, env); + expect((yield* suggestSplit.authorize(registry, other).pipe(Effect.flip))._tag).toBe( + "EnvironmentAuthorizationError", + ); expect((yield* mutate.authorize(registry, other).pipe(Effect.flip))._tag).toBe( "EnvironmentAuthorizationError", ); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index e0aae375ec5df..fe7a73c28d3b6 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -367,6 +367,10 @@ export function createThreadEnvironmentAtoms( `history:${JSON.stringify([environmentId, input.threadId])}`, }, }), + suggestSplit: createEnvironmentRpcCommand(runtime, { + label: "environment-data:commands:thread:suggest-split", + tag: WS_METHODS.orchestrationSuggestThreadSplit, + }), uploadFeedback: createEnvironmentRpcCommand(runtime, { label: "environment-data:commands:thread:upload-feedback", tag: WS_METHODS.providerUploadFeedback, diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index fafa54f5877e2..39f82d785eec8 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -17,6 +17,7 @@ import { WS_METHODS } from "./rpc.ts"; /** Incremental client enforcement; the server still authorizes every request. */ export const CLIENT_GUARDED_RPC_SCOPES = { + [WS_METHODS.orchestrationSuggestThreadSplit]: AuthOrchestrationOperateScope, [WS_METHODS.projectCodeQuery]: AuthFilesystemReadScope, [WS_METHODS.projectCodeFileHistory]: AuthOrchestrationReadScope, [WS_METHODS.projectKnowledgeMutate]: AuthOrchestrationOperateScope, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 0a8e0181598e9..96c046955cbe7 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -82,3 +82,5 @@ export * from "./projectKnowledge.ts"; export * from "./conversationSummaries.ts"; export * from "./projectLoom.ts"; + +export * from "./threadSplit.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index d5f41e0e67fca..a42950683f940 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1,3 +1,4 @@ +import { ThreadSplitSuggestionInput, ThreadSplitSuggestionResult } from "./threadSplit.ts"; import { ConversationSummaryDescribeInput, ConversationSummaryError, @@ -453,6 +454,7 @@ import { VcsError } from "./vcs.ts"; import { Project, ProjectMutation, ProjectMutationError } from "./project.ts"; export const WS_METHODS = { + orchestrationSuggestThreadSplit: "orchestration.suggestThreadSplit", projectCodeQuery: "projectCode.query", projectCodeFileHistory: "projectCode.fileHistory", projectCodeSubscribeIndex: "projectCode.subscribeIndex", @@ -2170,7 +2172,14 @@ export const WsProjectCodeRpcGroup = RpcGroup.make( WsProjectCodeSubscribeIndexRpc, ).middleware(RpcScopeAuthorization); +const WsThreadSplitSuggestionRpc = Rpc.make(WS_METHODS.orchestrationSuggestThreadSplit, { + payload: ThreadSplitSuggestionInput, + success: ThreadSplitSuggestionResult, + error: EnvironmentAuthorizationError, +}); + export const WsRpcGroup = RpcGroup.make( + WsThreadSplitSuggestionRpc, WsGuestShareCreateRpc, WsServerProbeRpc, WsServerGetConfigRpc, diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index f4ad157e61d5c..5a1895a3eb536 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1326,6 +1326,10 @@ export const ServerSettings = Schema.Struct({ snoozePresets: SnoozePresetSettings.pipe( Schema.withDecodingDefault(Effect.succeed(DEFAULT_SNOOZE_PRESETS)), ), + threadSplitSuggestionsEnabled: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(false)), + ), + threadSplitModelEndpoint: Schema.String.pipe(Schema.withDecodingDefault(Effect.succeed(""))), /** Beta: guest links for this machine's threads. Off refuses new links and pauses existing ones. */ guestSharingEnabled: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), /** @@ -1738,6 +1742,8 @@ export const ServerSettingsPatch = Schema.Struct({ autoResumeLimitedThreads: Schema.optionalKey(Schema.Boolean), snoozeLimitedThreads: Schema.optionalKey(Schema.Boolean), snoozePresets: Schema.optionalKey(SnoozePresetSettings), + threadSplitSuggestionsEnabled: Schema.optionalKey(Schema.Boolean), + threadSplitModelEndpoint: Schema.optionalKey(Schema.String), guestSharingEnabled: Schema.optionalKey(Schema.Boolean), linkedEnvironmentsEnabled: Schema.optionalKey(Schema.Boolean), backgroundActivity: Schema.optionalKey( diff --git a/packages/contracts/src/threadSplit.ts b/packages/contracts/src/threadSplit.ts new file mode 100644 index 0000000000000..722f9de082712 --- /dev/null +++ b/packages/contracts/src/threadSplit.ts @@ -0,0 +1,9 @@ +import * as Schema from "effect/Schema"; +import { ThreadId } from "./baseSchemas.ts"; +export const ThreadSplitSuggestionInput = Schema.Struct({ + threadId: ThreadId, + message: Schema.String.check(Schema.isMinLength(1), Schema.isMaxLength(12000)), +}); +export type ThreadSplitSuggestionInput = typeof ThreadSplitSuggestionInput.Type; +export const ThreadSplitSuggestionResult = Schema.Struct({ suggest: Schema.Boolean }); +export type ThreadSplitSuggestionResult = typeof ThreadSplitSuggestionResult.Type; diff --git a/packages/shared/package.json b/packages/shared/package.json index 42ccc86988910..1cf86cd0b6295 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -3,6 +3,7 @@ "private": true, "type": "module", "exports": { + "./threadSplitPrompt.json": "./src/threadSplitPrompt.json", "./pstack": { "types": "./src/pstack.ts", "import": "./src/pstack.ts" diff --git a/packages/shared/src/threadSplitPrompt.json b/packages/shared/src/threadSplitPrompt.json new file mode 100644 index 0000000000000..dedc0d18d4600 --- /dev/null +++ b/packages/shared/src/threadSplitPrompt.json @@ -0,0 +1,8 @@ +{ + "type": "noul", + "instructions": "Could the agent handle the new user message independently of all earlier thread context, without reusing any earlier result, file, decision, or instruction? Answer yes only if independent.", + "criteria": { + "true": "No earlier thread result, file, decision, or instruction is needed.", + "false": "Handling relies on an earlier thread result, file, decision, or instruction." + } +} From 278af1f1ffacbdfbaafa4844ce05d83defb42b50 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 14:07:25 -0800 Subject: [PATCH 3/7] fix(server): thread split suggestions ignore empty message context Most real user messages carry a context object with zero records, and the suggestion service treated any context object as a reason to abstain, so most threads could never receive a suggestion. Abstain only when a prior message actually references context records. Co-Authored-By: Claude Fable 5.1 --- .../ThreadSplitSuggestions.test.ts | 18 ++++++++++++++++-- .../orchestration-v2/ThreadSplitSuggestions.ts | 2 +- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSplitSuggestions.test.ts b/apps/server/src/orchestration-v2/ThreadSplitSuggestions.test.ts index 5806ae73630e9..567175badae28 100644 --- a/apps/server/src/orchestration-v2/ThreadSplitSuggestions.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSplitSuggestions.test.ts @@ -76,9 +76,19 @@ it.effect("abstains for disabled, unsafe endpoints, truncated replies and stale "wrong-model", "invalid", "stale", + "context", + "empty-context", "valid", ] as const) { const records = source(); + if (scenario === "context" || scenario === "empty-context") + records.messages[0] = { + ...records.messages[0]!, + context: { + version: 1, + records: scenario === "context" ? [{ kind: "file", contextId: "file-1" }] : [], + } as unknown as OrchestrationV2ConversationMessage["context"], + }; let requests = 0; const client = HttpClient.make((request) => Effect.sync(() => { @@ -123,8 +133,12 @@ it.effect("abstains for disabled, unsafe endpoints, truncated replies and stale const result = yield* Effect.gen(function* () { return yield* (yield* Suggestions.ThreadSplitSuggestions).suggest(input); }).pipe(Effect.provide(layer)); - assert.equal(result.suggest, scenario === "valid", scenario); - assert.equal(requests, scenario === "disabled" || scenario === "remote" ? 0 : 1); + assert.equal(result.suggest, scenario === "valid" || scenario === "empty-context", scenario); + assert.equal( + requests, + scenario === "disabled" || scenario === "remote" || scenario === "context" ? 0 : 1, + scenario, + ); } }), ); diff --git a/apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts b/apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts index 9763b74cab220..90227e1222757 100644 --- a/apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts +++ b/apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts @@ -86,7 +86,7 @@ const make = Effect.gen(function* () { if ( message.streaming || message.attachments.length || - message.context || + message.context?.records.length || message.notification || message.delegatedCompletion || (message.role === "user" && message.createdBy !== "user") || From 7680abbda9bb754273cb856eaaf448850ea00ca6 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 14:08:39 -0800 Subject: [PATCH 4/7] fix(chat): dismissed thread split suggestion stays hidden while the draft is edited Dismissal was keyed to the exact draft text, so continuing to type the same draft requested inference again and brought the suggestion back. Dismissal now holds until the draft is sent or cleared, on web and mobile. Mobile also skips requests when the environment has no decision endpoint, matching web. The user guide names the setting as the UI labels it. Co-Authored-By: Claude Fable 5.1 --- .../src/features/threads/ThreadComposer.tsx | 1 + .../features/threads/ThreadSplitSuggestion.tsx | 16 +++++++++------- .../chat/ThreadSplitSuggestion.browser.test.tsx | 8 +++++++- .../components/chat/ThreadSplitSuggestion.tsx | 11 ++++++----- docs/user/composer.md | 6 +++--- 5 files changed, 26 insertions(+), 16 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 949882f68c6e9..be9db25cc1942 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -917,6 +917,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps message={draftMessage} enabled={ props.serverConfig?.settings.threadSplitSuggestionsEnabled === true && + Boolean(props.serverConfig.settings.threadSplitModelEndpoint) && !props.isDraft && !props.hidden && !props.activeThreadBusy && diff --git a/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx b/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx index 62c3cdfc746c9..f009de3f07f7c 100644 --- a/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx +++ b/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx @@ -27,16 +27,22 @@ export function ThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { const canSuggest = useAtomValue( threadEnvironment.suggestSplit.permissionAtom(props.environmentId), ); - if (!props.enabled || !focused || !canSuggest || !props.message.trim()) return null; + const [dismissed, setDismissed] = useState(false); + const hasMessage = Boolean(props.message.trim()); + if (dismissed && !hasMessage) setDismissed(false); + if (!props.enabled || !focused || !canSuggest || !hasMessage || dismissed) return null; return ( setDismissed(true)} /> ); } -function ActiveThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { +function ActiveThreadSplitSuggestion( + props: ThreadSplitSuggestionProps & { readonly onDismiss: () => void }, +) { const navigation = useNavigation(); const [suggestedMessage, setSuggestedMessage] = useState(null); const suggest = useAtomCommand(threadEnvironment.suggestSplit, { @@ -86,11 +92,7 @@ function ActiveThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { setComposerDraftText(props.draftKey, ""); }} /> - + ), +})); +vi.mock("@effect/atom-react", () => ({ useAtomValue: () => true })); +vi.mock("../../state/threads", () => ({ + threadEnvironment: { suggestSplit: { permissionAtom: () => true } }, +})); +vi.mock("../../state/use-atom-command", () => ({ useAtomCommand: () => state.suggest })); +vi.mock("../../state/use-composer-drafts", () => ({ + createNewTaskDraft: () => "draft-new", + getComposerDraftSnapshot: () => ({ text: "", attachments: [], context: undefined }), + setComposerDraftText: vi.fn(), + updateComposerDraftSettings: vi.fn(), +})); + +import { ThreadSplitSuggestion } from "./ThreadSplitSuggestion"; + +const environmentId = EnvironmentId.make("environment-test"); +const threadId = ThreadId.make("thread-test"); +const projectId = ProjectId.make("project-test"); + +function mount(key: string, message: string) { + const container = document.createElement("div"); + const root = createRoot(container); + act(() => + root.render( + , + ), + ); + return { container, root }; +} + +describe("ThreadSplitSuggestion", () => { + it("keeps a dismissed suggestion hidden across remounts until the draft is cleared", async () => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + vi.useFakeTimers(); + try { + state.suggest.mockResolvedValue({ _tag: "Success", value: { suggest: true } }); + const first = mount("a", "Write release notes"); + await act(async () => { + await vi.advanceTimersByTimeAsync(1000); + }); + expect(state.suggest).toHaveBeenCalledTimes(1); + const dismiss = first.container.querySelector("button:last-of-type"); + expect(first.container.textContent).toContain("Start a new thread"); + act(() => dismiss?.dispatchEvent(new MouseEvent("click", { bubbles: true }))); + expect(first.container.textContent).not.toContain("Start a new thread"); + act(() => first.root.unmount()); + + const second = mount("b", "Write release notes"); + await act(async () => { + await vi.advanceTimersByTimeAsync(1000); + }); + expect(state.suggest).toHaveBeenCalledTimes(1); + expect(second.container.textContent).not.toContain("Start a new thread"); + act(() => second.root.unmount()); + + const cleared = mount("c", ""); + await act(async () => { + await vi.advanceTimersByTimeAsync(1000); + }); + act(() => cleared.root.unmount()); + + const fresh = mount("d", "A different draft"); + await act(async () => { + await vi.advanceTimersByTimeAsync(1000); + }); + expect(state.suggest).toHaveBeenCalledTimes(2); + await act(async () => { + await Promise.resolve(); + }); + expect(fresh.container.textContent).toContain("Start a new thread"); + act(() => fresh.root.unmount()); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx b/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx index f009de3f07f7c..10f55cfa929cd 100644 --- a/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx +++ b/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx @@ -12,6 +12,9 @@ import { setComposerDraftText, updateComposerDraftSettings, } from "../../state/use-composer-drafts"; +import { scopedThreadKey } from "../../lib/scopedEntities"; + +const dismissedThreadKeys = new Set(); type ThreadSplitSuggestionProps = { readonly environmentId: EnvironmentId; @@ -27,15 +30,22 @@ export function ThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { const canSuggest = useAtomValue( threadEnvironment.suggestSplit.permissionAtom(props.environmentId), ); - const [dismissed, setDismissed] = useState(false); + const threadKey = scopedThreadKey(props.environmentId, props.threadId); + const [dismissed, setDismissed] = useState(() => dismissedThreadKeys.has(threadKey)); const hasMessage = Boolean(props.message.trim()); - if (dismissed && !hasMessage) setDismissed(false); + if (dismissed && !hasMessage) { + dismissedThreadKeys.delete(threadKey); + setDismissed(false); + } if (!props.enabled || !focused || !canSuggest || !hasMessage || dismissed) return null; return ( setDismissed(true)} + onDismiss={() => { + dismissedThreadKeys.add(threadKey); + setDismissed(true); + }} /> ); } diff --git a/apps/mobile/vite.config.ts b/apps/mobile/vite.config.ts index 02c787bea771d..be46817dfc4d1 100644 --- a/apps/mobile/vite.config.ts +++ b/apps/mobile/vite.config.ts @@ -32,6 +32,7 @@ const reactTests = [ "src/features/settings/components/BranchNamingSettings.test.tsx", "src/features/settings/components/SettingsDragHandle.test.tsx", "src/features/threads/ThreadMarkdownImage.test.tsx", + "src/features/threads/ThreadSplitSuggestion.test.tsx", "src/features/threads/UsageLimitRecoveryCard.test.tsx", "src/features/threads/NewTaskDraftRouteScreen.test.tsx", "src/native/NativeHeaderToolbar.test.tsx", diff --git a/apps/web/src/components/chat/ThreadSplitSuggestion.browser.test.tsx b/apps/web/src/components/chat/ThreadSplitSuggestion.browser.test.tsx index dae5e21d8eb3d..2c6859d361b7b 100644 --- a/apps/web/src/components/chat/ThreadSplitSuggestion.browser.test.tsx +++ b/apps/web/src/components/chat/ThreadSplitSuggestion.browser.test.tsx @@ -67,8 +67,13 @@ function setup() { environmentId: EnvironmentId.make("environment-test"), threadId: ThreadId.make("thread-test"), }; - return () => ( - + return (key = "initial") => ( + ); } @@ -108,6 +113,10 @@ it("discards stale replies, supports opt-out and dismissal that outlasts edits, await screen.rerender(view()); await act(() => vi.advanceTimersByTime(1000)); expect(replies).toHaveLength(0); + await screen.rerender(view("remounted")); + await act(() => vi.advanceTimersByTime(1000)); + expect(replies).toHaveLength(0); + await expect.element(action).not.toBeInTheDocument(); state.source.prompt = ""; await screen.rerender(view()); diff --git a/apps/web/src/components/chat/ThreadSplitSuggestion.tsx b/apps/web/src/components/chat/ThreadSplitSuggestion.tsx index b317460d843f1..c6664898d2c92 100644 --- a/apps/web/src/components/chat/ThreadSplitSuggestion.tsx +++ b/apps/web/src/components/chat/ThreadSplitSuggestion.tsx @@ -2,7 +2,7 @@ import { useEffect, useEffectEvent, useState } from "react"; import { useAtomValue } from "@effect/atom-react"; import { AsyncResult } from "effect/reactivity"; import type { ProjectId, ScopedThreadRef } from "@t3tools/contracts"; -import { scopeProjectRef } from "@t3tools/client-runtime/environment"; +import { scopeProjectRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; import { useComposerDraftStore, composerDraftHasUserContent } from "~/composerDraftStore"; import { useEnvironmentSettings } from "~/hooks/useSettings"; import { useNewThreadHandler } from "~/hooks/useHandleNewThread"; @@ -11,6 +11,8 @@ import { threadEnvironment } from "~/state/threads"; import { toastManager } from "../ui/toastManager"; import { Button } from "../ui/button"; +const dismissedThreadKeys = new Set(); + export function ThreadSplitSuggestion({ threadRef, projectId, @@ -35,7 +37,8 @@ export function ThreadSplitSuggestion({ }); const newThread = useNewThreadHandler(); const [suggestedKey, setSuggestedKey] = useState(null); - const [dismissed, setDismissed] = useState(false); + const threadKey = scopedThreadKey(threadRef); + const [dismissed, setDismissed] = useState(() => dismissedThreadKeys.has(threadKey)); const [opening, setOpening] = useState(false); const enabled = settings.threadSplitSuggestionsEnabled && @@ -56,7 +59,10 @@ export function ThreadSplitSuggestion({ setObservedKey(selectionKey); setSuggestedKey(null); } - if (dismissed && !prompt.trim()) setDismissed(false); + if (dismissed && !prompt.trim()) { + dismissedThreadKeys.delete(threadKey); + setDismissed(false); + } const check = useEffectEvent(async (message: string, key: string, cancelled: () => boolean) => { const result = await suggest({ environmentId: threadRef.environmentId, @@ -124,7 +130,10 @@ export function ThreadSplitSuggestion({ type="button" variant="ghost" size="sm" - onClick={() => setDismissed(true)} + onClick={() => { + dismissedThreadKeys.add(threadKey); + setDismissed(true); + }} aria-label="Dismiss new thread suggestion" > Dismiss From 64ea876a6e4a337e8aee9b924dacde74019e0f23 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 14:30:52 -0800 Subject: [PATCH 6/7] chore(evals): ignore python bytecode cache Running the thread-split unittest leaves evals/thread-split/__pycache__ untracked; ignore it alongside eval run output. Co-Authored-By: SWE-2 High --- evals/.gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/evals/.gitignore b/evals/.gitignore index a1e03960fbab4..2fcb51f75d98c 100644 --- a/evals/.gitignore +++ b/evals/.gitignore @@ -1 +1,2 @@ runs/ +__pycache__/ From 7e06015821b8d2cf8f2b17ad7f8c2edcaf33342b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 3 Nov 2026 17:48:21 -0800 Subject: [PATCH 7/7] test(evals): measure thread split detection on held-out human conversations --- evals/thread-split/README.md | 100 +++++++ evals/thread-split/experiment.py | 404 ++++++++++++++++++++++++++ evals/thread-split/test_experiment.py | 100 +++++++ 3 files changed, 604 insertions(+) create mode 100644 evals/thread-split/experiment.py create mode 100644 evals/thread-split/test_experiment.py diff --git a/evals/thread-split/README.md b/evals/thread-split/README.md index 4db5a82862c7a..60e6105c02666 100644 --- a/evals/thread-split/README.md +++ b/evals/thread-split/README.md @@ -52,3 +52,103 @@ The app integration is opt-in, passive and asynchronous, using an explicitly configured local endpoint. It requires user action to move a draft into a fresh thread. Validate on new sessions before considering a default-on rollout. Automatic splitting and interrupting prompts are outside this feature. + +## Session-held-out experiment + +`experiment.py` separates score collection from labeling and calibration. It uses +three fixed yes/no prompts: the production baseline, an actionable-task/reference +check, and a contextual task check. Keep both packets and labels outside the repo. + +Packets contain `id`, `session`, `split` (`calibration` or `evaluation`), `state` +(as above), and `eligibility: {"eligible": true, "reasons": []}`. A session cannot +cross splits; `session_group` also cannot cross splits when supplied. Previously +exposed sessions are rejected from evaluation. The default minimum is 100 evaluation packets. An explicitly declared smaller +exploratory run can use `--min-evaluation-cases N` consistently on every command; +its minimum is bound into receipts and the frozen selection. Record the reason +before scoring and report the actual labeled and eligible denominators. Reducing +this minimum does not relax the calibration evidence requirements. Labels are separate +rows with `id` and `label`: `NEW_TASK`, `CONTINUE`, `NO_TASK`, or `UNCERTAIN`. +They should describe the usefulness of a fresh thread, not whether the observed +assistant happened to reuse history. Label before viewing scores. Preserve the +source provenance and independent-label agreement in the private corpus manifest. + +Freeze each split's final labels before scoring. Calculate its hash without sending +the labels to the scoring process (replace the path for each split): + +```sh +python3 -c 'import sys; sys.path.insert(0, "evals/thread-split"); from experiment import digest, labels; print(digest(labels(sys.argv[1])))' /private/cal-labels.json +``` + +```sh +python3 evals/thread-split/experiment.py score --packets /private/packets.json \ + --split calibration --labels-sha256 CALIBRATION_LABEL_HASH \ + --model /models/pplx-decider-flash-v1.2 --out /private/cal-scores +python3 evals/thread-split/experiment.py calibrate --packets /private/packets.json \ + --labels /private/cal-labels.json --scores /private/cal-scores/scores.jsonl \ + --out /private/selection.json +python3 evals/thread-split/experiment.py score --packets /private/packets.json \ + --split evaluation --selection /private/selection.json \ + --labels-sha256 EVALUATION_LABEL_HASH --model /models/pplx-decider-flash-v1.2 --out /private/test-scores +python3 evals/thread-split/experiment.py report --packets /private/packets.json \ + --split evaluation --selection /private/selection.json \ + --labels /private/test-labels.json --scores /private/test-scores/scores.jsonl \ + --out /private/report.json +``` + +Calibration refuses held-out labels or scores. It selects the prompt/cutoff with +most useful eligible detections subject to at most 1% observed false suggestions +among scored, eligible negative calibration cases. Selection requires at least +100 such negatives and five true detections. A candidate that fails on more +eligible cases than the baseline is disqualified. These are minimum evidence +requirements, not guarantees of generalisation. Ties prefer fewer false +suggestions, then the baseline prompt. If no useful cutoff qualifies, the frozen +threshold is null (no suggestions). This empirical constraint is not a population +guarantee. Evaluation scores only the frozen choice and the production baseline; +its report includes the baseline at the shipped 0.94 cutoff. + +The report retains every checkpoint in reconstructed-path recall, including +ineligible cases and abstentions; its false-suggestion rate describes all negative +checkpoints, including those the service would never score. `eligible_scored` is the +primary false-suggestion denominator used for calibration. `classifier` includes +all successful scores, even for ineligible packets. Uncertain labels remain visible but do not enter confusion counts. It also +reports class counts, predictions by class (including UNCERTAIN), eligibility +exclusions, failures, eligible-success latency median/p95 and +session-group-resampled intervals. Checkpoint-level Wilson bounds are also +reported with an explicit independence assumption; they do not account for +within-session correlation. In particular, a zero-width bootstrap interval at zero +observed errors must not be interpreted as certainty. Eligibility is supplied by corpus reconstruction, +not observed runtime behavior. Validate actual service availability, permissions, +busy-state checks and HTTP timeouts separately. Direct MLX scoring does not measure +those paths. Output files never overwrite existing results. An interrupted scoring run can use +`--resume` with identical arguments; corpus, driver, split, model path and prompts +must match its saved receipt. Receipts also bind model-file contents, MLX version, +label hashes and relevant source-file contents. Reports validate these receipts. +Evaluation creates a marker beside the selection file and can only resume in its +original output directory. Completed score rows are not run again. + +## Verified human follow-up evaluation + +The subsequent corpus contains 194 app-attributed human follow-ups across 24 +threads and five provider configurations. Independent judges agreed on 181 labels; +a third judge resolved 13 disagreements before scoring. Whole ancestry groups +separate 140 development/calibration cases from 54 held-out cases. This is one +user's history, not a representative population sample. + +At the shipped 0.94 threshold, the baseline produced no held-out suggestions. +Among 23 eligible cases, it missed the one actionable independent task and falsely +flagged none of 21 negatives; one case remained uncertain. Across all 54 cases, +there were three new tasks, two excluded by eligibility gates. Zero observed false +positives on 21 eligible negatives is weak evidence: the 95% Wilson upper bound is +15.5%, even assuming independent checkpoints. + +Replaying all 54 cases through the production service with reconstructed projection +fixtures and real local HTTP inference produced the same zero suggestions. It +requested inference for 23 cases and excluded 31, with no request-state mismatches +or observed failures. Sequential service latency was 494 ms median and 885 ms p95. +This exercises the service and HTTP path, not historical settings, authenticated +RPC, concurrent drafts, or client typing behavior. + +Three fixed prompts were scored on calibration data. Only 23 eligible negatives +were available, below the predeclared 100-negative minimum, so no tuned candidate +qualified. These results do not justify a threshold change or a readiness claim. +The feature remains experimental and opt-in. diff --git a/evals/thread-split/experiment.py b/evals/thread-split/experiment.py new file mode 100644 index 0000000000000..48d39029b3627 --- /dev/null +++ b/evals/thread-split/experiment.py @@ -0,0 +1,404 @@ +"""Session-held-out evaluation. Private packets and labels remain outside the repository.""" +import argparse +from collections import Counter, defaultdict +import hashlib +from importlib.metadata import version +import json +import math +from pathlib import Path +import random +import time + +from detector import QUESTION, score_response + +LABELS = {'NEW_TASK', 'CONTINUE', 'NO_TASK', 'UNCERTAIN'} +MIN_CALIBRATION_NEGATIVES = 100 +MIN_CALIBRATION_DETECTIONS = 5 +PROMPTS = { + 'baseline': QUESTION, + 'actionable': { + 'type': 'noul', + 'instructions': ( + 'Does the new user message request an actionable NEW task that can be handled in a fresh ' + 'thread without any earlier thread-specific information? Answer no for acknowledgements, ' + 'thanks, pauses, resumptions, confirmations, or progress checks. Answer no if resolving ' + 'references such as it, they, that, also, or the earlier work requires previous messages. ' + 'The fresh thread keeps the same project, environment, tools and standing project instructions. ' + 'A change of subject or the ability to rediscover facts does not by itself make a task independent.' + ), + 'criteria': {'true': 'Actionable new task; all necessary task-specific context is in the new message.', + 'false': 'Continuation, non-task, unresolved reference, or uncertain dependence.'}, + }, + 'contextual': { + 'type': 'noul', + 'instructions': ( + 'Should this message be offered a fresh thread? Check three conditions: it asks for ' + 'substantive work, it is a separate task rather than the next step of the current task, ' + 'and it can be understood without prior thread-specific entities, constraints, results, ' + 'or decisions. Answer yes only when all three hold. Greetings, acknowledgements, ' + 'pause/resume messages, status requests and ambiguous references are no. The fresh thread ' + 'keeps the same project, environment, tools and standing project instructions.' + ), + 'criteria': {'true': 'A self-contained actionable new task warrants a fresh thread.', + 'false': 'Earlier context matters, no new task is requested, or the distinction is unclear.'}, + }, +} + + +def digest(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False).encode()).hexdigest() + + +def read(path): + return json.loads(Path(path).read_text()) + + +def file_digest(path): + with Path(path).open('rb') as source: + return hashlib.file_digest(source, 'sha256').hexdigest() + + +def source_fingerprints(): + root = Path(__file__).resolve().parents[2] + paths = [Path(__file__).resolve(), Path(__file__).with_name('detector.py').resolve(), + root / 'packages/shared/src/threadSplitPrompt.json', + root / 'apps/server/src/orchestration-v2/ThreadSplitSuggestions.ts'] + return {str(path.relative_to(root)): file_digest(path) for path in paths} + + +def verify_receipt(receipt, artifact, split, truth=None): + for key, expected in artifact.items(): + if receipt.get(key) != expected: + raise ValueError(f'Score receipt mismatch: {key}') + if receipt.get('split') != split: + raise ValueError('Score receipt split mismatch') + for name, prompt in receipt['prompts'].items(): + if name not in PROMPTS or prompt != PROMPTS[name]: + raise ValueError('Score receipt prompt mismatch') + if truth is not None and receipt.get('labels_sha256') != digest(truth): + raise ValueError('Frozen labels mismatch') + + +def packets(path): + rows = read(path) + ids, sessions, groups = set(), {}, {} + for row in rows: + if row['id'] in ids: + raise ValueError('Duplicate packet id') + ids.add(row['id']) + split = row['split'] + if split not in ('calibration', 'evaluation'): + raise ValueError('Unknown split') + previous = sessions.setdefault(row['session'], split) + if previous != split: + raise ValueError('Session appears in calibration and evaluation') + group = row.get('session_group', row['session']) + if groups.setdefault(group, split) != split: + raise ValueError('Session group appears in calibration and evaluation') + if split == 'evaluation' and row.get('previously_exposed_session', False): + raise ValueError('Previously exposed session appears in evaluation') + eligibility = row['eligibility'] + if not isinstance(eligibility['eligible'], bool) or not isinstance(eligibility['reasons'], list): + raise ValueError('Eligibility must be explicit') + state = row['state'] + if set(state) != {'prior_thread', 'new_message'} or not isinstance(state['new_message'], str): + raise ValueError('Invalid model state') + for turn in state['prior_thread']: + if set(turn) != {'user', 'assistant_final'} or not all(isinstance(v, str) for v in turn.values()): + raise ValueError('Invalid prior turn') + return rows + + +def labels(path): + result = {} + for row in read(path): + if row['id'] in result or row['label'] not in LABELS: + raise ValueError('Duplicate or invalid label') + result[row['id']] = row['label'] + return result + + +def score_rows(path): + result = {} + for line in Path(path).read_text().splitlines(): + row = json.loads(line) + key = (row['id'], row['prompt']) + if key in result: + raise ValueError('Duplicate score') + score = row['score'] + if score is not None and (not math.isfinite(score) or not 0 <= score <= 1): + raise ValueError('Invalid score') + result[key] = row + return result + + +def ratio(a, b): + return a / b if b else None + + +def wilson_interval(successes, total): + if not total: + return None + z = 1.959963984540054 + proportion = successes / total + denominator = 1 + z * z / total + center = (proportion + z * z / (2 * total)) / denominator + margin = z * math.sqrt(proportion * (1 - proportion) / total + z * z / (4 * total * total)) / denominator + return [max(0, center - margin), min(1, center + margin)] + + +def metrics(rows, threshold, full_path): + counts = dict(tp=0, fp=0, tn=0, fn=0) + classes = Counter(row['label'] for row in rows) + abstained = 0 + predictions = Counter() + for row in rows: + usable = row['score'] is not None and (row['eligible'] or not full_path) + if usable and threshold is not None and row['score'] >= threshold: + predictions[row['label']] += 1 + if row['label'] == 'UNCERTAIN': + continue + usable = row['score'] is not None and (row['eligible'] or not full_path) + if not usable: + abstained += 1 + if not full_path: + continue + predicted = usable and threshold is not None and row['score'] >= threshold + positive = row['label'] == 'NEW_TASK' + counts[('t' if positive == predicted else 'f') + ('p' if predicted else 'n')] += 1 + tp, fp, tn, fn = (counts[k] for k in ('tp', 'fp', 'tn', 'fn')) + return dict(counts, total=len(rows), classes=dict(classes), abstained=abstained, + ineligible=sum(not r['eligible'] for r in rows), + failed=sum(r['eligible'] and r['score'] is None for r in rows), + predicted_by_class=dict(predictions), + binomial_wilson_ci95_assuming_independent_checkpoints=dict( + precision=wilson_interval(tp, tp + fp), recall=wilson_interval(tp, tp + fn), + false_suggestion_rate=wilson_interval(fp, fp + tn)), + precision=ratio(tp, tp + fp), recall=ratio(tp, tp + fn), + false_suggestion_rate=ratio(fp, fp + tn)) + + +def joined(data, truth, scores, prompt, split): + result = [] + for case in data: + if case['split'] != split: + continue + key = (case['id'], prompt) + if case['id'] not in truth or key not in scores: + raise ValueError('Every packet requires a label and a score/abstention receipt') + score = scores[key] + if score['session'] != case['session'] or score['state_sha256'] != digest(case['state']): + raise ValueError('Score does not belong to this packet state/session') + if score['prompt_sha256'] != digest(PROMPTS[prompt]): + raise ValueError('Prompt changed after scoring') + result.append(dict(score, label=truth[case['id']], eligible=case['eligibility']['eligible'], + session_group=case.get('session_group', case['session']))) + return result + + +def choose(rows, max_fpr): + rows = [r for r in rows if r['eligible'] and r['score'] is not None] + thresholds = sorted({r['score'] for r in rows if r['score'] is not None}, reverse=True) + choices = [] + for cutoff in thresholds: + m = metrics(rows, cutoff, False) + if (m['fp'] + m['tn'] >= MIN_CALIBRATION_NEGATIVES + and m['tp'] >= MIN_CALIBRATION_DETECTIONS + and m['false_suggestion_rate'] <= max_fpr): + choices.append((m['tp'], -m['fp'], cutoff)) + return max(choices)[2] if choices and max(choices)[0] > 0 else None + + +def percentile(values, fraction): + values = sorted(values) + if not values: + return None + index = (len(values) - 1) * fraction + lo = math.floor(index) + hi = math.ceil(index) + return values[lo] + (values[hi] - values[lo]) * (index - lo) + + +def session_intervals(rows, threshold, full_path): + groups = defaultdict(list) + for row in rows: + groups[row.get('session_group', row['session'])].append(row) + names = sorted(groups) + rng = random.Random(20261103) + draws = defaultdict(list) + for _ in range(1000): + sample = [row for name in rng.choices(names, k=len(names)) for row in groups[name]] + for key, value in metrics(sample, threshold, full_path).items(): + if key in ('precision', 'recall', 'false_suggestion_rate') and value is not None: + draws[key].append(value) + return {key: [percentile(values, .025), percentile(values, .975)] for key, values in draws.items()} + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('operation', choices=['score', 'calibrate', 'report']) + parser.add_argument('--packets', type=Path, required=True) + parser.add_argument('--out', type=Path, required=True) + parser.add_argument('--split', choices=['calibration', 'evaluation']) + parser.add_argument('--model', type=Path) + parser.add_argument('--resume', action='store_true') + parser.add_argument('--prompts', nargs='+', choices=list(PROMPTS), default=list(PROMPTS)) + parser.add_argument('--labels', type=Path) + parser.add_argument('--scores', type=Path) + parser.add_argument('--labels-sha256') + parser.add_argument('--selection', type=Path) + parser.add_argument('--max-fpr', type=float, default=.01) + parser.add_argument('--min-evaluation-cases', type=int, default=100) + args = parser.parse_args() + data = packets(args.packets) + if args.min_evaluation_cases < 1: + parser.error('--min-evaluation-cases must be positive') + if sum(c['split'] == 'evaluation' for c in data) < args.min_evaluation_cases: + raise ValueError('Insufficient session-held-out evaluation packets for declared minimum') + artifact = {'packets_sha256': digest(data), 'source_sha256': source_fingerprints(), + 'minimum_evaluation_cases': args.min_evaluation_cases} + if args.operation == 'score': + if not args.model or not args.split: + parser.error('score requires --model and --split') + if not args.labels_sha256 or len(args.labels_sha256) != 64 or any(c not in '0123456789abcdef' for c in args.labels_sha256): + parser.error('score requires --labels-sha256 of the frozen id-to-label mapping') + model_files = {str(p.relative_to(args.model)): file_digest(p) + for p in sorted(args.model.rglob('*')) + if p.is_file() and p.suffix in ('.json', '.safetensors', '.model')} + if not model_files: + raise ValueError('No model files found') + model_identity = dict(files=model_files, mlx_decision_version=version('mlx-decision')) + selected = args.prompts + if args.split == 'evaluation': + if not args.selection: + parser.error('evaluation scoring requires frozen --selection') + selection = read(args.selection) + if selection['packets_sha256'] != digest(data) or selection['minimum_evaluation_cases'] != args.min_evaluation_cases: + raise ValueError('Corpus changed after calibration') + if selection['source_sha256'] != artifact['source_sha256'] or selection['model_identity'] != model_identity: + raise ValueError('Source or model changed after calibration') + selected = list(dict.fromkeys(['baseline', selection['prompt']])) + if selection['prompt_sha256'] != digest(PROMPTS[selection['prompt']]): + raise ValueError('Selected prompt changed') + receipt = dict(artifact, model=str(args.model), split=args.split, + prompts={p: PROMPTS[p] for p in selected}, + model_identity=model_identity, labels_sha256=args.labels_sha256) + if args.split == 'evaluation': + receipt['selection_sha256'] = digest(selection) + marker = args.selection.with_suffix(args.selection.suffix + '.evaluated') + if args.resume: + if read(marker) != str(args.out.resolve()): + raise ValueError('Evaluation resume directory changed') + else: + with marker.open('x') as output: + json.dump(str(args.out.resolve()), output) + if args.resume: + if read(args.out / 'receipt.json') != receipt: + raise ValueError('Run metadata changed; cannot resume') + existing = score_rows(args.out / 'scores.jsonl') if (args.out / 'scores.jsonl').exists() else {} + else: + args.out.mkdir(parents=True, exist_ok=False) + (args.out / 'receipt.json').write_text(json.dumps(receipt, indent=2)) + existing = {} + import mlx_decision + model = mlx_decision.load(str(args.model)) + with (args.out / 'scores.jsonl').open('a' if args.resume else 'x') as output: + for prompt in selected: + for case in data: + if case['split'] != args.split: + continue + if (case['id'], prompt) in existing: + continue + row = dict(id=case['id'], session=case['session'], split=args.split, + prompt=prompt, prompt_sha256=digest(PROMPTS[prompt]), + state_sha256=digest(case['state']), score=None, seconds=None) + # Score bounded model inputs even when a product eligibility guard would abstain. + if len(json.dumps(case['state'], ensure_ascii=False, separators=(',', ':')).encode('utf-16-le')) // 2 > 24000: + row['abstention'] = 'state_limit' + else: + start = time.perf_counter() + try: + response = model.decide_request({'state': case['state'], 'questions': {'q': PROMPTS[prompt]}}) + row['score'] = score_response(response) + if row['score'] is None: + row['abstention'] = 'truncated' + except Exception as error: + row['abstention'] = type(error).__name__ + row['seconds'] = time.perf_counter() - start + output.write(json.dumps(row) + '\n') + output.flush() + print(prompt, args.split, 'complete', flush=True) + return + if not args.labels or not args.scores: + parser.error('calibrate/report require --labels and --scores') + truth, scores = labels(args.labels), score_rows(args.scores) + split = 'calibration' if args.operation == 'calibrate' else args.split + receipt = read(args.scores.with_name('receipt.json')) + verify_receipt(receipt, artifact, split, truth) + artifact.update(model_identity=receipt['model_identity'], labels_sha256=digest(truth), + scores_sha256=file_digest(args.scores)) + if args.operation == 'calibrate': + if not 0 <= args.max_fpr <= 1: + raise ValueError('Invalid FPR budget') + calibration_ids = {c['id'] for c in data if c['split'] == 'calibration'} + if any(case_id not in calibration_ids for case_id, _ in scores): + raise ValueError('Calibration cannot read evaluation or unknown scores') + if any(case_id not in calibration_ids for case_id in truth): + raise ValueError('Calibration requires calibration-only labels') + baseline = joined(data, truth, scores, 'baseline', 'calibration') + baseline_failures = sum(r['eligible'] and r['score'] is None for r in baseline) + candidates = [] + for prompt in args.prompts: + rows = joined(data, truth, scores, prompt, 'calibration') + failures = sum(r['eligible'] and r['score'] is None for r in rows) + threshold = choose(rows, args.max_fpr) if failures <= baseline_failures else None + candidates.append(dict(prompt=prompt, threshold=threshold, + exceeds_baseline_failures=failures > baseline_failures, metrics=metrics(rows, threshold, True), + classifier=metrics(rows, threshold, False), + eligible_scored=metrics([r for r in rows if r['eligible']], threshold, False))) + winner = max(candidates, key=lambda c: (c['metrics']['tp'], -c['metrics']['fp'], c['prompt'] == 'baseline')) + artifact.update(prompt=winner['prompt'], threshold=winner['threshold'], + prompt_sha256=digest(PROMPTS[winner['prompt']]), candidates=candidates, + calibration_sessions=sorted({c['session'] for c in data if c['split'] == 'calibration'}), + labels_sha256=digest({c['id']: truth[c['id']] for c in data if c['split'] == 'calibration'}), + minimum_scored_negative_cases=MIN_CALIBRATION_NEGATIVES, + minimum_true_detections=MIN_CALIBRATION_DETECTIONS, + max_false_suggestion_rate=args.max_fpr) + else: + if not args.selection or not args.split: + parser.error('report requires frozen --selection and --split') + selection = read(args.selection) + if selection['packets_sha256'] != digest(data) or selection['minimum_evaluation_cases'] != args.min_evaluation_cases: + raise ValueError('Corpus changed after calibration') + if selection['source_sha256'] != artifact['source_sha256'] or selection['model_identity'] != artifact['model_identity']: + raise ValueError('Source or model changed after calibration') + if args.split == 'evaluation' and receipt.get('selection_sha256') != digest(selection): + raise ValueError('Selection changed after evaluation') + rows = joined(data, truth, scores, selection['prompt'], args.split) + baseline_rows = joined(data, truth, scores, 'baseline', args.split) + threshold = selection['threshold'] + if threshold is not None and not 0 <= threshold <= 1: + raise ValueError('Invalid frozen threshold') + elapsed = [r['seconds'] for r in rows if r['eligible'] and r['score'] is not None and r['seconds'] is not None] + artifact.update(split=args.split, prompt=selection['prompt'], threshold=threshold, + selection_sha256=digest(selection), sessions=len({r['session'] for r in rows}), + eligibility_scope='reconstructed_static; runtime availability, busy state, permissions and HTTP timeout require separate validation', + classifier=metrics(rows, threshold, False), + eligible_scored=metrics([r for r in rows if r['eligible']], threshold, False), + reconstructed_path=metrics(rows, threshold, True), + baseline=dict(threshold=.94, classifier=metrics(baseline_rows, .94, False), + eligible_scored=metrics([r for r in baseline_rows if r['eligible']], .94, False), + reconstructed_path=metrics(baseline_rows, .94, True)), + classifier_ci95=session_intervals(rows, threshold, False), + reconstructed_path_ci95=session_intervals(rows, threshold, True), + eligible=sum(r['eligible'] for r in rows), + exclusions=dict(Counter(reason for c in data if c['split'] == args.split for reason in c['eligibility']['reasons'])), + failures=dict(Counter(r.get('abstention') for r in rows if r['score'] is None)), + latency_seconds=dict(median=percentile(elapsed, .5), p95=percentile(elapsed, .95))) + with args.out.open('x') as output: + json.dump(artifact, output, indent=2) + print(json.dumps(artifact)) + + +if __name__ == '__main__': + main() diff --git a/evals/thread-split/test_experiment.py b/evals/thread-split/test_experiment.py new file mode 100644 index 0000000000000..381a60a36b052 --- /dev/null +++ b/evals/thread-split/test_experiment.py @@ -0,0 +1,100 @@ +import json +from pathlib import Path +import tempfile +import unittest + +from experiment import choose, joined, metrics, packets, digest, PROMPTS, wilson_interval, verify_receipt + + +class ExperimentTest(unittest.TestCase): + def test_abstention_and_ineligible_switches_remain_in_path_recall(self): + rows = [ + dict(id='a', label='NEW_TASK', score=.9, eligible=True), + dict(id='b', label='NEW_TASK', score=.9, eligible=False), + dict(id='c', label='NEW_TASK', score=None, eligible=True), + dict(id='d', label='NO_TASK', score=.95, eligible=True), + dict(id='e', label='CONTINUE', score=.1, eligible=True), + dict(id='f', label='UNCERTAIN', score=.99, eligible=True), + ] + classifier = metrics(rows, .8, False) + path = metrics(rows, .8, True) + self.assertEqual((classifier['tp'], classifier['fn'], classifier['abstained']), (2, 0, 1)) + self.assertEqual((path['tp'], path['fn'], path['abstained']), (1, 2, 2)) + self.assertEqual(path['recall'], 1 / 3) + self.assertEqual(path['fp'], 1) + self.assertEqual(path['classes']['UNCERTAIN'], 1) + eligible = metrics([r for r in rows if r['eligible']], .8, False) + self.assertEqual(eligible['false_suggestion_rate'], .5) + self.assertEqual(eligible['predicted_by_class']['UNCERTAIN'], 1) + self.assertEqual(path['ineligible'], 1) + self.assertEqual(path['failed'], 1) + self.assertGreater(wilson_interval(0, 100)[1], .03) + self.assertLess(wilson_interval(3, 3)[0], .5) + self.assertIsNone(wilson_interval(0, 0)) + + def test_no_admissible_useful_cutoff_disables_instead_of_relaxing_budget(self): + rows = [dict(label='NEW_TASK', score=.8, eligible=True), + dict(label='NO_TASK', score=.9, eligible=True)] + self.assertIsNone(choose(rows, .01)) + self.assertIsNone(choose(rows, 1)) + enough = ([dict(label='NEW_TASK', score=.8, eligible=True) for _ in range(5)] + + [dict(label='CONTINUE', score=.1, eligible=True) for _ in range(100)]) + self.assertEqual(choose(enough, .01), .8) + enough[-1]['score'] = None + self.assertIsNone(choose(enough, .01)) + + def test_session_leakage_and_receipt_drift_are_rejected(self): + state = dict(prior_thread=[dict(user='Run tests', assistant_final='Passed')], new_message='Thanks') + data = [dict(id='a', session='same', split='calibration', state=state, + eligibility=dict(eligible=True, reasons=[])), + dict(id='b', session='same', split='evaluation', state=state, + eligibility=dict(eligible=True, reasons=[]))] + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'packets.json' + path.write_text(json.dumps(data)) + with self.assertRaisesRegex(ValueError, 'Session appears'): + packets(path) + data[1]['session'] = 'different-session-same-ancestry' + for row in data: + row['session_group'] = 'shared-ancestry' + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'packets.json' + path.write_text(json.dumps(data)) + with self.assertRaisesRegex(ValueError, 'Session group appears'): + packets(path) + data[1]['session_group'] = 'unseen-group' + data[1]['previously_exposed_session'] = True + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'packets.json' + path.write_text(json.dumps(data)) + with self.assertRaisesRegex(ValueError, 'Previously exposed'): + packets(path) + receipt = dict(id='a', session='same', state_sha256=digest(state), + prompt_sha256=digest(PROMPTS['baseline']), score=.1) + rows = joined(data[:1], {'a': 'NO_TASK'}, {('a', 'baseline'): receipt}, 'baseline', 'calibration') + self.assertEqual(len(rows), 1) + receipt['state_sha256'] = 'wrong' + with self.assertRaisesRegex(ValueError, 'packet state'): + joined(data[:1], {'a': 'NO_TASK'}, {('a', 'baseline'): receipt}, 'baseline', 'calibration') + with self.assertRaisesRegex(ValueError, 'Every packet'): + joined(data[:1], {'a': 'NO_TASK'}, {}, 'baseline', 'calibration') + + def test_frozen_receipt_rejects_changed_labels_sources_and_split(self): + artifact = dict(packets_sha256='packet-digest', source_sha256={'detector.py': 'source-digest'}) + truth = {'a': 'CONTINUE'} + receipt = dict(artifact, split='evaluation', prompts={'baseline': PROMPTS['baseline']}, + labels_sha256=digest(truth)) + verify_receipt(receipt, artifact, 'evaluation', truth) + with self.assertRaisesRegex(ValueError, 'Frozen labels'): + verify_receipt(receipt, artifact, 'evaluation', {'a': 'NEW_TASK'}) + with self.assertRaisesRegex(ValueError, 'source_sha256'): + verify_receipt(receipt, dict(artifact, source_sha256={'detector.py': 'new'}), 'evaluation', truth) + with self.assertRaisesRegex(ValueError, 'split mismatch'): + verify_receipt(receipt, artifact, 'calibration', truth) + receipt['prompts'] = {'baseline': {'instructions': 'changed'}} + with self.assertRaisesRegex(ValueError, 'prompt mismatch'): + verify_receipt(receipt, artifact, 'evaluation', truth) + + +if __name__ == '__main__': + unittest.main()