fix(pi): validate user-entry conversation forks

electron/coding-runtime/conversation-service.ts

@@ -491,7 +491,27 @@ export class CodingConversationService {
   }
 
   async fork(sourceConversationId: string, sourceEntryId?: string): Promise<CodingConversationV2> {
+    const entryId = requiredString(sourceEntryId, 'Fork source entry id', 256);
     const prepared = await this.ensurePrepared(sourceConversationId);
+    let sourceSnapshot: ConversationSnapshot;
+    try {
+      sourceSnapshot = await this.runtime.getSnapshot(sourceConversationId);
+    } catch (error) {
+      runtimeError(error);
+    }
+    const sourceNode = sourceSnapshot.nodes.find((node) => (
+      node.kind === 'message'
+      && node.sourceEntryId === entryId
+      && node.role === 'user'
+      && node.status !== 'optimistic'
+    ));
+    if (!sourceNode) {
+      throw new CodingConversationServiceError(
+        400,
+        'CODING_CONVERSATION_REQUEST_INVALID',
+        'Fork source must be a durable user message on the active Conversation path',
+      );
+    }
     const source = await this.projects.findActiveConversation(sourceConversationId);
     const store = this.projects.conversationStore(source.project.path);
     const created = await persist(() => store.create({
@@ -503,7 +523,7 @@ export class CodingConversationService {
     try {
       await this.runtime.fork({
         sourceConversationId,
-        ...(sourceEntryId ? { sourceEntryId } : {}),
+        sourceEntryId: entryId,
         conversation: {
           ...prepared,
           conversationId: created.id,

electron/coding-runtime/pi/release-proof.ts

@@ -104,10 +104,12 @@ type ProxyCompositionRun = {
   provider: LocalProofProvider;
   projectId: string;
   projectPath: string;
+  sourceConversationId?: string;
   hostToken: string;
   originalAcceptPrompt: CodingProductComposition['conversations']['acceptPrompt'];
   submissionInjection: { definiteRejections: number };
   activeStatus?: PiReleaseProxyCompositionStatus;
+  settledStatus?: PiReleaseProxyCompositionStatus;
 };
 
 type ResilienceCompositionRun = {
@@ -212,6 +214,18 @@ export interface PiReleaseProxyCompositionProof extends PiReleaseProxyCompositio
     definiteRejectionObserved: boolean;
     retryAccepted: boolean;
   };
+  fork: {
+    conversationCount: number;
+    sameAgent: boolean;
+    targetBindingEstablished: boolean;
+    distinctSessionBinding: boolean;
+    workerStatus: string;
+    runStatus: string;
+    sourceEntryRole: 'user';
+    targetHydratedBeforeSourceEntry: boolean;
+    sourceUnchanged: boolean;
+    promptReplayObserved: boolean;
+  };
   currentHostTokenUsedBy: ProofWorkerRole[];
   released: { workers: number };
 }
@@ -916,8 +930,12 @@ async function proxyCompositionStatus(
     .filter(({ role }) => role === 'child')
     .map(({ processId }) => processId)
     .sort((left, right) => left - right);
+  const conversations = await run.composition.conversations.listConversations(run.projectId);
+  if (!run.sourceConversationId && conversations.length === 1) {
+    run.sourceConversationId = conversations[0]?.id;
+  }
   return {
-    conversationCount: (await run.composition.conversations.listConversations(run.projectId)).length,
+    conversationCount: conversations.length,
     providerRequests,
     activeProviderRequests: run.provider.activeCounts(),
     workers: diagnostics.workers,
@@ -1442,6 +1460,12 @@ export async function getFinalAsarProxyCompositionStatus(): Promise<PiReleasePro
     && status.processes.parent.length > 0) {
     run.activeStatus = status;
   }
+  if (status.conversationCount === 1
+    && status.activeProviderRequests.parent === 0
+    && status.activeProviderRequests.child === 0
+    && status.providerCompatibility.systemRoleRequests >= 2) {
+    run.settledStatus = status;
+  }
   return status;
 }
 
@@ -1456,10 +1480,12 @@ export async function finishFinalAsarProxyCompositionProof(): Promise<PiReleaseP
   if (!run) throw new Error('PI proxy composition proof is not running');
   try {
     const conversations = await run.composition.conversations.listConversations(run.projectId);
-    if (conversations.length !== 1) {
-      throw new Error(`Expected one first proxy Conversation, received ${conversations.length}`);
+    if (conversations.length !== 2 || !run.sourceConversationId) {
+      throw new Error(`Expected source and forked proxy Conversations, received ${conversations.length}`);
     }
-    const conversation = conversations[0]!;
+    const conversation = conversations.find(({ id }) => id === run.sourceConversationId);
+    const forked = conversations.find(({ id }) => id !== run.sourceConversationId);
+    if (!conversation || !forked) throw new Error('Proxy fork source or target Conversation is unavailable');
     let snapshot = await run.composition.conversations.getSnapshot(conversation.id);
     const deadline = Date.now() + 30_000;
     while (Date.now() < deadline && (
@@ -1473,9 +1499,24 @@ export async function finishFinalAsarProxyCompositionProof(): Promise<PiReleaseP
     if (snapshot.run.status !== 'idle' || !serializedNodes.includes('REAL_PARENT_COMPLETE')) {
       throw new Error('First proxy Conversation did not settle with the controlled parent response');
     }
-    const binding = await run.composition.projects.conversationStore(run.projectPath).get(conversation.id);
+    const forkedSnapshot = await run.composition.conversations.getSnapshot(forked.id);
+    const sourceEntry = snapshot.nodes.find((node) => (
+      node.kind === 'message'
+      && node.role === 'user'
+      && node.blocks.some((block) => block.kind === 'text'
+        && block.text.includes('Retry the packaged Main proxy Conversation'))
+    ));
+    if (!sourceEntry || sourceEntry.kind !== 'message' || !sourceEntry.sourceEntryId) {
+      throw new Error('Proxy source user entry is not durable after the controlled turn');
+    }
+    const [binding, forkedBinding] = await Promise.all([
+      run.composition.projects.conversationStore(run.projectPath).get(conversation.id),
+      run.composition.projects.conversationStore(run.projectPath).get(forked.id),
+    ]);
     const finalStatus = await proxyCompositionStatus(run);
     const activeStatus = run.activeStatus ?? finalStatus;
+    const settledStatus = run.settledStatus;
+    if (!settledStatus) throw new Error('Proxy source Conversation did not record a settled pre-fork baseline');
     const currentHostTokenUsedBy: ProofWorkerRole[] = [
       ...(activeStatus.providerRequests.parent > 0 ? ['parent' as const] : []),
       ...(activeStatus.providerRequests.child > 0 ? ['child' as const] : []),
@@ -1486,6 +1527,27 @@ export async function finishFinalAsarProxyCompositionProof(): Promise<PiReleaseP
     if (!binding?.piSessionId || !binding.sessionKey) {
       throw new Error('First proxy Conversation did not persist its Pi session binding');
     }
+    if (!forkedBinding?.piSessionId || !forkedBinding.sessionKey) {
+      throw new Error('Forked proxy Conversation did not persist its Pi session binding');
+    }
+    const distinctSessionBinding = binding.piSessionId !== forkedBinding.piSessionId
+      && binding.sessionKey !== forkedBinding.sessionKey;
+    const targetHydratedBeforeSourceEntry = forkedSnapshot.nodes.every((node) => (
+      node.kind !== 'message' || node.sourceEntryId !== sourceEntry.sourceEntryId
+    )) && !JSON.stringify(forkedSnapshot.nodes).includes('REAL_PARENT_COMPLETE');
+    if (forked.agentId !== conversation.agentId
+      || !distinctSessionBinding
+      || forkedSnapshot.worker.status !== 'ready'
+      || forkedSnapshot.run.status !== 'idle'
+      || !targetHydratedBeforeSourceEntry) {
+      throw new Error(`Proxy user-entry fork proof failed: ${JSON.stringify({
+        sameAgent: forked.agentId === conversation.agentId,
+        distinctSessionBinding,
+        workerStatus: forkedSnapshot.worker.status,
+        runStatus: forkedSnapshot.run.status,
+        targetHydratedBeforeSourceEntry,
+      })}`);
+    }
     if (!finalStatus.providerCompatibility.roleContract.developerRejected
       || !finalStatus.providerCompatibility.roleContract.systemAccepted
       || finalStatus.providerCompatibility.developerRoleRequests !== 0
@@ -1512,7 +1574,7 @@ export async function finishFinalAsarProxyCompositionProof(): Promise<PiReleaseP
       tokenSafety,
       providerMode: 'loopback-through-authenticated-host-proxy',
       firstConversation: {
-        count: conversations.length,
+        count: 1,
         bindingEstablished: true,
         workerStatus: snapshot.worker.status,
         runStatus: snapshot.run.status,
@@ -1521,6 +1583,20 @@ export async function finishFinalAsarProxyCompositionProof(): Promise<PiReleaseP
         definiteRejectionObserved: finalStatus.providerCompatibility.controlledDefiniteRejections === 1,
         retryAccepted: finalStatus.providerCompatibility.systemRoleRequests >= 2,
       },
+      fork: {
+        conversationCount: conversations.length,
+        sameAgent: forked.agentId === conversation.agentId,
+        targetBindingEstablished: true,
+        distinctSessionBinding,
+        workerStatus: forkedSnapshot.worker.status,
+        runStatus: forkedSnapshot.run.status,
+        sourceEntryRole: 'user',
+        targetHydratedBeforeSourceEntry,
+        sourceUnchanged: serializedNodes.includes('Retry the packaged Main proxy Conversation')
+          && serializedNodes.includes('REAL_PARENT_COMPLETE'),
+        promptReplayObserved: finalStatus.providerRequests.parent !== settledStatus.providerRequests.parent
+          || finalStatus.providerRequests.child !== settledStatus.providerRequests.child,
+      },
       currentHostTokenUsedBy,
     };
     await cleanupProxyCompositionRun(run);