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 f8bfb4fd0725d..53313dcdf3087 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,28 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps /> ) : null} + + ({ + suggest: vi.fn(), +})); + +vi.mock("react-native", () => ({ + View: ({ children }: { children: ReactNode }) =>
{children}
, +})); +vi.mock("@react-navigation/native", () => ({ + StackActions: { push: (name: string, params: unknown) => ({ name, params }) }, + useIsFocused: () => true, + useNavigation: () => ({ dispatch: vi.fn() }), +})); +vi.mock("../../components/ui/Button", () => ({ + Button: ({ label, onPress }: { label: string; onPress: () => void }) => ( + + ), +})); +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 new file mode 100644 index 0000000000000..10f55cfa929cd --- /dev/null +++ b/apps/mobile/src/features/threads/ThreadSplitSuggestion.tsx @@ -0,0 +1,108 @@ +import { useEffect, useEffectEvent, useState } from "react"; +import { StackActions, useIsFocused, useNavigation } from "@react-navigation/native"; +import type { EnvironmentId, ProjectId, ThreadId } from "@t3tools/contracts"; +import { View } from "react-native"; +import { Button } from "../../components/ui/Button"; +import { threadEnvironment } from "../../state/threads"; +import { useAtomValue } from "@effect/atom-react"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { + createNewTaskDraft, + getComposerDraftSnapshot, + setComposerDraftText, + updateComposerDraftSettings, +} from "../../state/use-composer-drafts"; +import { scopedThreadKey } from "../../lib/scopedEntities"; + +const dismissedThreadKeys = new Set(); + +type ThreadSplitSuggestionProps = { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly draftKey: string; + readonly message: string; + readonly enabled: boolean; +}; + +export function ThreadSplitSuggestion(props: ThreadSplitSuggestionProps) { + const focused = useIsFocused(); + const canSuggest = useAtomValue( + threadEnvironment.suggestSplit.permissionAtom(props.environmentId), + ); + const threadKey = scopedThreadKey(props.environmentId, props.threadId); + const [dismissed, setDismissed] = useState(() => dismissedThreadKeys.has(threadKey)); + const hasMessage = Boolean(props.message.trim()); + if (dismissed && !hasMessage) { + dismissedThreadKeys.delete(threadKey); + setDismissed(false); + } + if (!props.enabled || !focused || !canSuggest || !hasMessage || dismissed) return null; + return ( + { + dismissedThreadKeys.add(threadKey); + setDismissed(true); + }} + /> + ); +} + +function ActiveThreadSplitSuggestion( + props: ThreadSplitSuggestionProps & { readonly onDismiss: () => void }, +) { + 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 bfb86dd7250d0..99976ab798f2d 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 a72904eed2fe2..6ffa6d35ed319 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -458,3 +458,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 **Suggest new threads** 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; +it stays hidden until you send or clear that draft. 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/.gitignore b/evals/.gitignore index a1e03960fbab4..2fcb51f75d98c 100644 --- a/evals/.gitignore +++ b/evals/.gitignore @@ -1 +1,2 @@ runs/ +__pycache__/ 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..60e6105c02666 --- /dev/null +++ b/evals/thread-split/README.md @@ -0,0 +1,154 @@ +# 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 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. + +## 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. + +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/detector.py b/evals/thread-split/detector.py new file mode 100644 index 0000000000000..faafe0049e52a --- /dev/null +++ b/evals/thread-split/detector.py @@ -0,0 +1,45 @@ +import math + +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': QUESTION}} + + +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/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/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() 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() 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 88a5759c5ed50..1684db9b3f25e 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 1eda82aab0e1d..e68df0582d489 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -83,3 +83,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 4af79f3396ce4..acc0b1b1563d1 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, @@ -454,6 +455,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", @@ -2180,7 +2182,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 7d2b635a6acbc..f9aef396c0b49 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1327,6 +1327,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(""))), /** Clients hide threads with these tags while `hideTaggedThreads` is on. Bad entries drop alone, so a hand edit cannot reset settings. */ hiddenThreadTags: Schema.Array(Schema.Unknown).pipe( Schema.decodeTo( @@ -1754,6 +1758,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), hiddenThreadTags: Schema.optionalKey(Schema.Array(Schema.String)), hideTaggedThreads: Schema.optionalKey(Schema.Boolean), guestSharingEnabled: Schema.optionalKey(Schema.Boolean), 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." + } +}