feat: route MR events and authenticated implementation results

crates/server-api/src/lib.rs

@@ -366,6 +366,8 @@ impl_openapi_schema!(
     TicketAssignmentsPage,
     TicketAssignmentAnswerRequest,
     TicketAssignmentAnswerResponse,
+    TicketImplementationResultRequest,
+    TicketImplementationResultReceipt,
     TicketAssignmentDiagnosticsQuery,
     TicketAssignmentDiagnosticsPage,
     TicketQueryRequest,
@@ -3277,6 +3279,33 @@ pub trait ServerApi {
         #[body] request: TicketAssignmentAnswerRequest,
     ) -> Result<TicketAssignmentAnswerResponse, RepositoryApiError>;
 
+    #[post("/workspaces/{workspace_id}/tickets/{ticket_id}/implementation-results", status = 200, error_status = 400, additional_error_statuses = [401, 403, 404, 409, 422, 500], bearer_auth = true, browser_auth = true)]
+    async fn ticket_implementation_result(
+        &self,
+        #[extension] context: ServerRequestContext,
+        #[path] workspace_id: String,
+        #[path] ticket_id: String,
+        #[body] request: TicketImplementationResultRequest,
+    ) -> Result<TicketImplementationResultReceipt, RepositoryApiError>;
+
+    #[post("/api/w/{workspace_id}/tickets/{ticket_id}/implementation-results", status = 200, error_status = 400, additional_error_statuses = [401, 403, 404, 409, 422, 500], bearer_auth = true, browser_auth = true)]
+    async fn ticket_implementation_result_browser(
+        &self,
+        #[extension] context: ServerRequestContext,
+        #[path] workspace_id: String,
+        #[path] ticket_id: String,
+        #[body] request: TicketImplementationResultRequest,
+    ) -> Result<TicketImplementationResultReceipt, RepositoryApiError>;
+
+    #[get("/workspaces/{workspace_id}/tickets/{ticket_id}/implementation-results/{result_id}", status = 200, error_status = 400, additional_error_statuses = [401, 403, 404, 500], bearer_auth = true, browser_auth = true)]
+    async fn ticket_implementation_result_get(
+        &self,
+        #[extension] context: ServerRequestContext,
+        #[path] workspace_id: String,
+        #[path] ticket_id: String,
+        #[path] result_id: String,
+    ) -> Result<TicketImplementationResultReceipt, RepositoryApiError>;
+
     #[get("/workspaces/{workspace_id}/tickets/{ticket_id}/assignment-tracking", status = 200, error_status = 400, additional_error_statuses = [401, 403, 404, 500], bearer_auth = true, browser_auth = true)]
     async fn ticket_assignment_tracking_get(
         &self,
@@ -8006,6 +8035,34 @@ transparent_ticket_dto!(
     ticket::TicketAssignmentAnswerReceipt
 );
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum TicketImplementationResultKind {
+    Handoff,
+    Result,
+    Blocker,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct TicketImplementationResultRequest {
+    pub operation_key: String,
+    pub assignment_id: String,
+    pub orchestrator_assignment_id: String,
+    pub kind: TicketImplementationResultKind,
+    pub body: String,
+    pub merge_request: Option<MergeRequestReviewSubject>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct TicketImplementationResultReceipt {
+    pub result_id: String,
+    pub ticket_id: String,
+    pub request: TicketImplementationResultRequest,
+    pub created_at: String,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
 #[serde(deny_unknown_fields)]
 pub struct TicketAssignmentDiagnosticsQuery {

crates/server-api/src/repository_openapi_typescript.rs

@@ -543,6 +543,27 @@ fn validate_ticket_browser_operations(
             request: None,
             successes: &[("200", "TicketAssignmentDiagnosticsPage")],
         },
+        OperationContract {
+            path: "/workspaces/{workspace_id}/tickets/{ticket_id}/implementation-results",
+            method: "post",
+            operation_id: "ticket_implementation_result",
+            request: Some("TicketImplementationResultRequest"),
+            successes: &[("200", "TicketImplementationResultReceipt")],
+        },
+        OperationContract {
+            path: "/api/w/{workspace_id}/tickets/{ticket_id}/implementation-results",
+            method: "post",
+            operation_id: "ticket_implementation_result_browser",
+            request: Some("TicketImplementationResultRequest"),
+            successes: &[("200", "TicketImplementationResultReceipt")],
+        },
+        OperationContract {
+            path: "/workspaces/{workspace_id}/tickets/{ticket_id}/implementation-results/{result_id}",
+            method: "get",
+            operation_id: "ticket_implementation_result_get",
+            request: None,
+            successes: &[("200", "TicketImplementationResultReceipt")],
+        },
         OperationContract {
             path: "/workspaces/{workspace_id}/tickets/{ticket_id}/assignment-answer",
             method: "post",

crates/worker-runtime/src/worker_source.rs

@@ -655,6 +655,26 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
                                     .await,
                             )
                         }
+                        WorkspaceServerOperation::TicketImplementationResult {
+                            ticket_id,
+                            request,
+                        } => generated_workspace_response(
+                            client
+                                .ticket_implementation_result(workspace_id, ticket_id, request)
+                                .await,
+                        ),
+                        WorkspaceServerOperation::TicketImplementationResultGet {
+                            ticket_id,
+                            result_id,
+                        } => generated_workspace_response(
+                            client
+                                .ticket_implementation_result_get(
+                                    workspace_id,
+                                    ticket_id,
+                                    result_id,
+                                )
+                                .await,
+                        ),
                         WorkspaceServerOperation::WorkerControlList => {
                             generated_workspace_response(
                                 client.worker_control_list(workspace_id).await,

crates/worker/src/feature/builtin/ticket.rs

@@ -639,7 +639,10 @@ impl TicketFeatureAccess {
                         assignment_public::READ_TOOL_NAMES.contains(name)
                             || self.authoring
                             || self.workflow
-                            || (*name == "TicketAssignmentAnswer" && self.thread)
+                            || (matches!(
+                                *name,
+                                "TicketAssignmentAnswer" | "TicketImplementationResult"
+                            ) && self.thread)
                     }),
             )
             .collect()
@@ -665,6 +668,7 @@ const READ_ONLY_TOOL_NAMES: &[&str] = &[
     "TicketAssignmentTrackingHistory",
     "TicketAssignmentsQuery",
     "TicketAssignmentDiagnostics",
+    "TicketImplementationResultQuery",
 ];
 
 const AUTHORING_TOOL_NAMES: &[&str] = &[
@@ -676,7 +680,7 @@ const AUTHORING_TOOL_NAMES: &[&str] = &[
     "TicketRelationRemove",
 ];
 
-const THREAD_TOOL_NAMES: &[&str] = &["TicketComment"];
+const THREAD_TOOL_NAMES: &[&str] = &["TicketComment", "TicketImplementationResult"];
 
 #[cfg(test)]
 const WORKSPACE_AUTHORING_TOOL_NAMES: &[&str] = &[
@@ -701,6 +705,8 @@ const WORKSPACE_AUTHORING_TOOL_NAMES: &[&str] = &[
     "TicketAssignmentsQuery",
     "TicketAssignmentDiagnostics",
     "TicketAssignmentAnswer",
+    "TicketImplementationResult",
+    "TicketImplementationResultQuery",
 ];
 
 #[cfg(test)]
@@ -727,6 +733,8 @@ const WORKFLOW_TOOL_NAMES: &[&str] = &[
     "TicketAssignmentsQuery",
     "TicketAssignmentDiagnostics",
     "TicketAssignmentAnswer",
+    "TicketImplementationResult",
+    "TicketImplementationResultQuery",
 ];
 
 const WORKFLOW_ADDITIONAL_TOOL_NAMES: &[&str] = &[
@@ -1637,6 +1645,18 @@ fn native_ticket_operation(tool_name: &str) -> Option<NativeTicketOperation> {
             identity_field: Some("ticket"),
             mutating: false,
         },
+        "TicketImplementationResult" => NativeTicketOperation {
+            operation: "implementation_result",
+            surface: NativeTicketSurface::Item,
+            identity_field: Some("ticket"),
+            mutating: true,
+        },
+        "TicketImplementationResultQuery" => NativeTicketOperation {
+            operation: "implementation_result_query",
+            surface: NativeTicketSurface::Item,
+            identity_field: Some("ticket"),
+            mutating: false,
+        },
         "TicketAssignmentAnswer" => NativeTicketOperation {
             operation: "assignment_answer",
             surface: NativeTicketSurface::Item,
@@ -1765,6 +1785,8 @@ pub(crate) fn enabled_ticket_definitions(
         assignment_public::PublicAssignmentKind::Assignments,
         assignment_public::PublicAssignmentKind::Diagnostics,
         assignment_public::PublicAssignmentKind::Answer,
+        assignment_public::PublicAssignmentKind::ImplementationResult,
+        assignment_public::PublicAssignmentKind::ImplementationResultQuery,
     ] {
         if allowed.contains(&kind.name()) {
             definitions.push(assignment_public::definition(client.clone(), kind));

crates/worker/src/feature/builtin/ticket/assignment_public.rs

@@ -22,9 +22,14 @@ pub(super) const TOOL_NAMES: &[&str] = &[
     "TicketAssignmentsQuery",
     "TicketAssignmentDiagnostics",
     "TicketAssignmentAnswer",
+    "TicketImplementationResult",
+    "TicketImplementationResultQuery",
+];
+pub(super) const READ_TOOL_NAMES: &[&str] = &[
+    "TicketAssignmentsQuery",
+    "TicketAssignmentDiagnostics",
+    "TicketImplementationResultQuery",
 ];
-pub(super) const READ_TOOL_NAMES: &[&str] =
-    &["TicketAssignmentsQuery", "TicketAssignmentDiagnostics"];
 
 #[derive(Clone, Copy)]
 pub(super) enum PublicAssignmentKind {
@@ -33,6 +38,8 @@ pub(super) enum PublicAssignmentKind {
     Assignments,
     Diagnostics,
     Answer,
+    ImplementationResult,
+    ImplementationResultQuery,
 }
 
 #[derive(Deserialize, JsonSchema)]
@@ -80,6 +87,20 @@ struct AnswerInput {
     request: server_api::TicketAssignmentAnswerRequest,
 }
 
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct ImplementationResultInput {
+    ticket: String,
+    #[serde(flatten)]
+    request: server_api::TicketImplementationResultRequest,
+}
+#[derive(Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct ImplementationResultQuery {
+    ticket: String,
+    result_id: String,
+}
+
 impl PublicAssignmentKind {
     pub(super) fn name(self) -> &'static str {
         match self {
@@ -88,6 +109,8 @@ impl PublicAssignmentKind {
             Self::Assignments => TOOL_NAMES[2],
             Self::Diagnostics => TOOL_NAMES[3],
             Self::Answer => TOOL_NAMES[4],
+            Self::ImplementationResult => TOOL_NAMES[5],
+            Self::ImplementationResultQuery => TOOL_NAMES[6],
         }
     }
     fn schema(self) -> Value {
@@ -109,6 +132,10 @@ impl PublicAssignmentKind {
             Self::Assignments => json!(schemars::schema_for!(AssignmentsInput)),
             Self::Diagnostics => json!(schemars::schema_for!(DiagnosticsInput)),
             Self::Answer => json!(schemars::schema_for!(AnswerInput)),
+            Self::ImplementationResult => json!(schemars::schema_for!(ImplementationResultInput)),
+            Self::ImplementationResultQuery => {
+                json!(schemars::schema_for!(ImplementationResultQuery))
+            }
         }
     }
 }
@@ -129,6 +156,12 @@ pub(super) fn description(name: &str) -> &'static str {
         "TicketAssignmentAnswer" => {
             "Append continuing input to the observed current waiting/inprogress assignment with assignment_id, expected_last_event_id, stable operation_key and bounded answer body. Query current tracking first. Answers continue the same row without automatic inprogress; done requires explicit new Assign/reissue instead. Backend authenticates/fences source, current assignment and event CAS, then saves input/audit/exact receipt atomically and notifies committed input. Exact retry retains full input/key; stale/cleared/replaced conflicts require rereading, and unknown outcome must not be retried as a fresh operation. No model actor/role claims, new assignment, case closure or Worker start."
         }
+        "TicketImplementationResult" => {
+            "Save an explicit handoff, result or blocker for the current implementation assignment and its original Orchestrator delegation. Read ShowTicket first for assignment identities. Backend authenticates the implementation Worker and fences both assignments; this is not AssignmentAnswer or a handling update. Optional MR/source identifies retained evidence, never approval. Retry an unknown response using exactly the same operation_key and full input; changed input with the same key is rejected. Result and durable notification source commit together. No case closure, new Assign, Worker stop or integration occurs."
+        }
+        "TicketImplementationResultQuery" => {
+            "Read one saved implementation result by Ticket and result_id from a result notification/receipt. Historical evidence is not a new request or approval; reread current delegation and MR before acting."
+        }
         _ => unreachable!("unknown public assignment tool"),
     }
 }
@@ -245,6 +278,47 @@ impl Tool for PublicAssignmentTool {
                     query: input.query,
                 }
             }
+            PublicAssignmentKind::ImplementationResult => {
+                let input: ImplementationResultInput =
+                    serde_json::from_str(input).map_err(parse)?;
+                identity("ticket", &input.ticket)?;
+                for (name, value) in [
+                    ("assignment_id", &input.request.assignment_id),
+                    (
+                        "orchestrator_assignment_id",
+                        &input.request.orchestrator_assignment_id,
+                    ),
+                    ("operation_key", &input.request.operation_key),
+                ] {
+                    identity(name, value)?;
+                }
+                if input.request.body.trim().is_empty()
+                    || input.request.body.len() > 16384
+                    || input.request.body.contains('\0')
+                {
+                    return Err(ToolError::InvalidArgument(
+                        "Result body must be nonblank, NUL-free and at most 16384 bytes".into(),
+                    ));
+                }
+                if let Some(mr) = &input.request.merge_request {
+                    identity("merge_request_id", &mr.merge_request_id)?;
+                    identity("subject_ref", &mr.subject_ref)?;
+                }
+                WorkspaceServerOperation::TicketImplementationResult {
+                    ticket_id: input.ticket,
+                    request: input.request,
+                }
+            }
+            PublicAssignmentKind::ImplementationResultQuery => {
+                let input: ImplementationResultQuery =
+                    serde_json::from_str(input).map_err(parse)?;
+                identity("ticket", &input.ticket)?;
+                identity("result_id", &input.result_id)?;
+                WorkspaceServerOperation::TicketImplementationResultGet {
+                    ticket_id: input.ticket,
+                    result_id: input.result_id,
+                }
+            }
             PublicAssignmentKind::Answer => {
                 let input: AnswerInput = serde_json::from_str(input).map_err(parse)?;
                 identity("ticket", &input.ticket)?;
@@ -320,6 +394,13 @@ impl Tool for PublicAssignmentTool {
                 }
                 serde_json::to_value(result)
             }
+            PublicAssignmentKind::ImplementationResult
+            | PublicAssignmentKind::ImplementationResultQuery => serde_json::to_value(
+                serde_json::from_str::<server_api::TicketImplementationResultReceipt>(
+                    &response.body,
+                )
+                .map_err(decode)?,
+            ),
             PublicAssignmentKind::Answer => serde_json::to_value(
                 serde_json::from_str::<server_api::TicketAssignmentAnswerResponse>(&response.body)
                     .map_err(decode)?,

crates/worker/src/feature/builtin/ticket/assignment_public/tests.rs

@@ -57,6 +57,16 @@ fn cases() -> Vec<(PublicAssignmentKind, Value, Value)> {
             json!({"ticket":"T-743", "assignment_id":"delegation", "after_input_event_id":"ticket-event:7", "after_start_operation_id":"start/7", "limit":10}),
             json!({"assignment_id":"delegation", "current":true, "input_origin_known":true, "actionable":true, "inputs":[], "next_input_event_id":null, "starts":[], "next_start_operation_id":null}),
         ),
+        (
+            PublicAssignmentKind::ImplementationResult,
+            json!({"ticket":"T-743","assignment_id":"implementation","orchestrator_assignment_id":"delegation","operation_key":"result-1","kind":"result","body":"Research without an MR","merge_request":null}),
+            json!({"result_id":"result-id","ticket_id":"T-743","created_at":"now","request":{"assignment_id":"implementation","orchestrator_assignment_id":"delegation","operation_key":"result-1","kind":"result","body":"Research without an MR","merge_request":null}}),
+        ),
+        (
+            PublicAssignmentKind::ImplementationResultQuery,
+            json!({"ticket":"T-743","result_id":"result-id"}),
+            json!({"result_id":"result-id","ticket_id":"T-743","created_at":"now","request":{"assignment_id":"implementation","orchestrator_assignment_id":"delegation","operation_key":"result-1","kind":"result","body":"Research without an MR","merge_request":null}}),
+        ),
         (
             PublicAssignmentKind::Answer,
             json!({"ticket":"T-743", "assignment_id":"delegation", "expected_last_event_id":"waiting", "operation_key":"answer-1", "body":"Proceed with option A"}),

crates/worker/src/worker.rs

@@ -754,6 +754,14 @@ pub enum WorkspaceServerOperation {
         ticket_id: String,
         request: server_api::TicketAssignmentAnswerRequest,
     },
+    TicketImplementationResult {
+        ticket_id: String,
+        request: server_api::TicketImplementationResultRequest,
+    },
+    TicketImplementationResultGet {
+        ticket_id: String,
+        result_id: String,
+    },
     WorkerControlList,
     WorkerControlSpawn(server_api::CreateWorkspaceWorkerRequest),
     WorkerControlInput {
@@ -925,6 +933,23 @@ fn workspace_server_operation_request(
                 &request,
             )
         }
+        WorkspaceServerOperation::TicketImplementationResult { ticket_id, request } => {
+            workspace_server_json_request(
+                format!(
+                    "/workspaces/{workspace_id}/tickets/{}/implementation-results",
+                    encode_workspace_path_segment(&ticket_id)
+                ),
+                &request,
+            )
+        }
+        WorkspaceServerOperation::TicketImplementationResultGet {
+            ticket_id,
+            result_id,
+        } => Ok(WorkspaceRequest::get(format!(
+            "/workspaces/{workspace_id}/tickets/{}/implementation-results/{}",
+            encode_workspace_path_segment(&ticket_id),
+            encode_workspace_path_segment(&result_id)
+        ))),
         WorkspaceServerOperation::WorkerControlList => Ok(WorkspaceRequest::get(format!(
             "{base}/worker-control/workers"
         ))),

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -469,7 +469,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
                     (
                         95,
                         "Common notification routing and durable Worker delivery".into()
-                    )
+                    ),
+                    (96, "MR notifications and authenticated implementation results".into())
                 ]
             );
             for (table, rows) in retained.iter().zip(original.iter()) {

crates/workspace-server/src/implementation_result.rs

@@ -0,0 +1,83 @@
+//! Implementation results are authenticated facts, not Orchestrator answers or handling updates.
+//! The write lock fences the original implementation assignment and its pinned delegation.
+//! Replays return the original result after reassignment, but cannot authorize a new result.
+use super::*;
+
+impl SqliteWorkspaceStore {
+    pub(crate) fn implementation_result(
+        &self,
+        workspace: &str,
+        ticket: &str,
+        result: &str,
+    ) -> Result<server_api::TicketImplementationResultReceipt> {
+        self.with_conn(|conn| {
+            let saved: Option<(String,String)> = conn.query_row("SELECT request_json,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND ticket_id=?2 AND result_id=?3",params![workspace,ticket,result],|r| Ok((r.get(0)?,r.get(1)?))).optional()?;
+            let (request,created_at) = saved.ok_or_else(|| Error::InvalidInput("Implementation result not found in this Ticket".into()))?;
+            Ok(server_api::TicketImplementationResultReceipt { result_id: result.into(), ticket_id: ticket.into(), request: serde_json::from_str(&request).map_err(|e| Error::Store(e.to_string()))?, created_at })
+        })
+    }
+    pub(crate) fn record_implementation_result(
+        &self,
+        workspace: &str,
+        ticket: &str,
+        request: &server_api::TicketImplementationResultRequest,
+        worker: &RuntimeWorkerRef,
+        binding: &str,
+    ) -> Result<server_api::TicketImplementationResultReceipt> {
+        for (name, value) in [
+            ("operation_key", &request.operation_key),
+            ("assignment_id", &request.assignment_id),
+            (
+                "orchestrator_assignment_id",
+                &request.orchestrator_assignment_id,
+            ),
+        ] {
+            if value.trim().is_empty() || value.len() > 1024 || value.chars().any(char::is_control)
+            {
+                return Err(Error::InvalidInput(format!(
+                    "{name} must be nonempty, control-free and at most 1024 bytes"
+                )));
+            }
+        }
+        if request.body.trim().is_empty()
+            || request.body.len() > 16384
+            || request.body.contains('\0')
+        {
+            return Err(Error::InvalidInput(
+                "Result body must be nonblank, NUL-free and at most 16384 bytes".into(),
+            ));
+        }
+        let encoded = serde_json::to_string(request).map_err(|e| Error::Store(e.to_string()))?;
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let authenticated: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2 AND binding_id=?3 AND revoked_at IS NULL)", params![workspace,worker.runtime_id,binding], |r| r.get(0))?;
+            if !authenticated {
+                return Err(Error::WorkspacePermissionDenied("Implementation result requires its authenticated Runtime binding".into()));
+            }
+            let saved: Option<(String,String,String,String,String,String)> = tx.query_row("SELECT result_id,ticket_id,request_json,runtime_id,worker_id,created_at FROM ticket_implementation_results WHERE workspace_id=?1 AND operation_key=?2",params![workspace,request.operation_key], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
+            if let Some((result_id,saved_ticket,saved_request,runtime,source,created_at)) = saved {
+                if saved_ticket != ticket || saved_request != encoded || runtime != worker.runtime_id || source != worker.worker_id {
+                    return Err(Error::TicketAssignmentConflict("Implementation result operation key was used for different input or source".into()));
+                }
+                return Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at });
+            }
+            let owns: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments c JOIN ticket_implementation_delegations d ON d.workspace_id=c.workspace_id AND d.assignment_id=c.assignment_id WHERE c.workspace_id=?1 AND c.ticket_id=?2 AND c.role='worker' AND c.assignment_id=?3 AND c.principal_kind='worker' AND c.runtime_id=?4 AND c.worker_id=?5 AND d.orchestrator_assignment_id=?6 AND NOT EXISTS(SELECT 1 FROM ticket_assignment_work_releases r WHERE r.workspace_id=c.workspace_id AND r.assignment_id=c.assignment_id))",params![workspace,ticket,request.assignment_id,worker.runtime_id,worker.worker_id,request.orchestrator_assignment_id],|r|r.get(0))?;
+            if !owns {
+                return Err(Error::WorkspacePermissionDenied("Result source is not the current implementation Worker for this delegation; legacy unknown work requires explicit assignment confirmation".into()));
+            }
+            // A delayed new result is rejected rather than retargeted to a new Orchestrator.
+            notification_routing::require_current(&tx,workspace,ticket,&request.orchestrator_assignment_id,"orchestrator")?;
+            if let Some(mr) = &request.merge_request {
+                // References identify retained evidence, not a model's assertion that a branch
+                // is approved/current. Integration must resolve Git and reread the latest MR.
+                let valid: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_ticket_relations r JOIN merge_request_thread_events e ON e.workspace_id=r.workspace_id AND e.merge_request_id=r.merge_request_id WHERE r.workspace_id=?1 AND r.ticket_id=?2 AND r.merge_request_id=?3 AND (json_extract(e.payload_json,'$.subject_ref')=?4 OR json_extract(e.payload_json,'$.approved_source_ref')=?4))",params![workspace,ticket,mr.merge_request_id,mr.subject_ref],|r|r.get(0))?;
+                if !valid { return Err(Error::InvalidInput("MR/source reference does not identify retained evidence for this Ticket".into())); }
+            }
+            let result_id = uuid::Uuid::now_v7().to_string();
+            let created_at = chrono::Utc::now().to_rfc3339();
+            tx.execute("INSERT INTO ticket_implementation_results VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10)",params![workspace,ticket,request.operation_key,result_id,request.assignment_id,request.orchestrator_assignment_id,worker.runtime_id,worker.worker_id,encoded,created_at])?;
+            tx.commit()?;
+            Ok(server_api::TicketImplementationResultReceipt { result_id, ticket_id: ticket.into(), request: request.clone(), created_at })
+        })
+    }
+}

crates/workspace-server/src/implementation_result.sql

@@ -0,0 +1,51 @@
+CREATE TABLE merge_request_source_observations (
+ sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+ workspace_id TEXT NOT NULL, merge_request_id TEXT NOT NULL,
+ subject_ref TEXT NOT NULL, previous_subject_ref TEXT,
+ observed_at TEXT NOT NULL,
+ FOREIGN KEY(workspace_id,merge_request_id) REFERENCES merge_requests(workspace_id,merge_request_id)
+);
+CREATE INDEX mr_source_observation_latest ON merge_request_source_observations(workspace_id,merge_request_id,sequence);
+CREATE TRIGGER mr_source_observations_no_update BEFORE UPDATE ON merge_request_source_observations BEGIN SELECT RAISE(ABORT,'Source observations are immutable'); END;
+CREATE TRIGGER mr_source_observations_no_delete BEFORE DELETE ON merge_request_source_observations BEGIN SELECT RAISE(ABORT,'Source observations are immutable'); END;
+INSERT INTO merge_request_source_observations(workspace_id,merge_request_id,subject_ref,observed_at)
+ SELECT e.workspace_id,e.merge_request_id,json_extract(e.payload_json,'$.subject_ref'),e.created_at
+ FROM merge_request_thread_events e WHERE e.sequence=(SELECT MAX(x.sequence) FROM merge_request_thread_events x WHERE x.workspace_id=e.workspace_id AND x.merge_request_id=e.merge_request_id AND json_extract(x.payload_json,'$.subject_ref') IS NOT NULL);
+
+-- Delegation is pinned when implementation responsibility is created, not at result arrival.
+CREATE TABLE ticket_implementation_delegations (
+ workspace_id TEXT NOT NULL, assignment_id TEXT NOT NULL, orchestrator_assignment_id TEXT,
+ PRIMARY KEY(workspace_id,assignment_id),
+ FOREIGN KEY(workspace_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id),
+ FOREIGN KEY(workspace_id,orchestrator_assignment_id) REFERENCES ticket_worker_assignments(workspace_id,assignment_id)
+);
+CREATE TRIGGER implementation_delegation_capture AFTER INSERT ON ticket_worker_assignments
+WHEN NEW.role='worker' BEGIN
+ INSERT INTO ticket_implementation_delegations
+ SELECT NEW.workspace_id,NEW.assignment_id,c.assignment_id FROM (SELECT 1) seed
+ LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=NEW.workspace_id AND c.ticket_id=NEW.ticket_id AND c.role='orchestrator';
+END;
+-- Retain a pre-upgrade parent only when its committed startup operation has exactly
+-- one authenticated delegation fence. Never infer ownership from time or comment prose.
+-- Older ambiguous rows remain unknown; explicit reassignment can confirm that work.
+INSERT INTO ticket_implementation_delegations
+ SELECT w.workspace_id,w.assignment_id,
+ (SELECT MIN(f.orchestrator_assignment_id) FROM ticket_assignment_operations o
+  JOIN ticket_assignment_operation_fences f ON f.workspace_id=o.workspace_id AND f.operation_id=o.operation_id
+  JOIN ticket_worker_assignments a ON a.workspace_id=f.workspace_id AND a.assignment_id=f.orchestrator_assignment_id AND a.ticket_id=w.ticket_id AND a.role='orchestrator'
+  WHERE o.workspace_id=w.workspace_id AND o.assignment_id=w.assignment_id AND o.claim_state='committed'
+  HAVING COUNT(DISTINCT f.orchestrator_assignment_id)=1)
+ FROM ticket_worker_assignments w WHERE w.role='worker';
+CREATE TABLE ticket_implementation_results (
+ workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, operation_key TEXT NOT NULL,
+ result_id TEXT NOT NULL, assignment_id TEXT NOT NULL, orchestrator_assignment_id TEXT NOT NULL,
+ runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, request_json TEXT NOT NULL CHECK(json_valid(request_json)),
+ created_at TEXT NOT NULL,
+ PRIMARY KEY(workspace_id,operation_key), UNIQUE(workspace_id,result_id),
+ FOREIGN KEY(workspace_id,ticket_id,assignment_id) REFERENCES ticket_worker_assignments(workspace_id,ticket_id,assignment_id),
+ FOREIGN KEY(workspace_id,ticket_id,orchestrator_assignment_id) REFERENCES ticket_worker_assignments(workspace_id,ticket_id,assignment_id)
+);
+CREATE TRIGGER implementation_results_no_update BEFORE UPDATE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
+CREATE TRIGGER implementation_results_no_delete BEFORE DELETE ON ticket_implementation_results BEGIN SELECT RAISE(ABORT,'Implementation results are immutable'); END;
+CREATE TRIGGER implementation_delegations_no_update BEFORE UPDATE ON ticket_implementation_delegations BEGIN SELECT RAISE(ABORT,'Implementation delegation is immutable'); END;
+CREATE TRIGGER implementation_delegations_no_delete BEFORE DELETE ON ticket_implementation_delegations BEGIN SELECT RAISE(ABORT,'Implementation delegation is immutable'); END;

crates/workspace-server/src/implementation_result_tests.rs

@@ -0,0 +1,310 @@
+// Production SQLite/store tests; signed HTTP and persistent Worker delivery are tested
+// in server_assignment_delivery_tests. These do not substitute for caller authentication.
+fn implementation_result_input() -> server_api::TicketImplementationResultRequest {
+    server_api::TicketImplementationResultRequest {
+        operation_key: "result-operation".into(),
+        assignment_id: "implementation".into(),
+        orchestrator_assignment_id: "delegation".into(),
+        kind: server_api::TicketImplementationResultKind::Result,
+        body: "Research complete; no repository was required".into(),
+        merge_request: None,
+    }
+}
+
+#[tokio::test]
+async fn implementation_result_replay_is_atomic_and_cannot_be_retargeted() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let input = implementation_result_input();
+    let before = store
+        .get_ticket_assignment_tracking("w", "tracking", "delegation")
+        .unwrap();
+    let receipt = store
+        .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+        .unwrap();
+    assert_eq!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .unwrap(),
+        receipt
+    );
+    assert_eq!(
+        store
+            .implementation_result("w", "tracking", &receipt.result_id)
+            .unwrap(),
+        receipt
+    );
+    assert!(
+        store
+            .implementation_result("w", "other", &receipt.result_id)
+            .is_err()
+    );
+    assert_eq!(
+        store
+            .get_ticket_assignment_tracking("w", "tracking", "delegation")
+            .unwrap(),
+        before
+    );
+    let mut changed = input.clone();
+    changed.body.push('!');
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    for (workspace, ticket, source, binding) in [
+        ("other", "tracking", worker.clone(), "tracking-binding"),
+        ("w", "other", worker.clone(), "tracking-binding"),
+        (
+            "w",
+            "tracking",
+            RuntimeWorkerRef::new("r", "worker"),
+            "tracking-binding",
+        ),
+        ("w", "tracking", worker.clone(), "wrong-binding"),
+    ] {
+        assert!(
+            store
+                .record_implementation_result(workspace, ticket, &input, &source, binding)
+                .is_err()
+        );
+    }
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            Ok(())
+        })
+        .unwrap();
+    let pending = store
+        .list_pending_ticket_assignment_deliveries("w", 100)
+        .unwrap();
+    assert!(
+        pending
+            .iter()
+            .any(|p| p.input_event_id == format!("implementation-result:{}", receipt.result_id))
+    );
+    // Reassignment does not rewrite a receipt or deliver old results as new work.
+    tracking_store_assign(&store, "replacement-delegation", "replace");
+    assert_eq!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .unwrap(),
+        receipt
+    );
+    changed.operation_key = "late-result".into();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    changed.orchestrator_assignment_id = "replacement-delegation".into();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &changed, &worker, "tracking-binding")
+            .is_err()
+    );
+    assert!(
+        !store
+            .list_pending_ticket_assignment_deliveries("w", 100)
+            .unwrap()
+            .iter()
+            .any(|p| p.input_event_id.starts_with("implementation-result:"))
+    );
+}
+
+#[tokio::test]
+async fn implementation_result_rejects_forged_evidence_and_rolls_back_without_notification() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let mut input = implementation_result_input();
+    input.merge_request = Some(server_api::MergeRequestReviewSubject {
+        merge_request_id: "fake".into(),
+        subject_ref: "fake".into(),
+    });
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .is_err()
+    );
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+#[tokio::test]
+async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknown_legacy_results() {
+    let (_dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    routing_worker_assign(&store, "implementation");
+    routing_worker_assign(&store, "known-implementation");
+    store.with_conn(|conn| {
+        conn.execute("INSERT INTO ticket_assignment_operation_fences VALUES('w','operation-known-implementation','delegation','r','worker','workspace-orchestrator','tracking-binding',?1)",[serde_json::to_string(&tracking_store_source()).unwrap()])?;
+        // Reconstruct the prior composition before its new business capture existed.
+        conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER implementation_delegation_capture;
+            DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations;
+            UPDATE __yoi_schema_migrations SET version=95 WHERE version=96;")?;
+        conn.execute_batch("INSERT INTO merge_requests VALUES('w','historical-mr','main','open','refs/heads/work','refs/heads/main','3','3');
+            INSERT INTO merge_request_ticket_relations VALUES('w','historical-mr','tracking','implements','3');
+            INSERT INTO merge_request_thread_events VALUES('w','historical-mr','approved-event',1,'review','{\"subject_ref\":\"source-a\",\"decision\":\"approve\"}',NULL,'3');
+            INSERT INTO merge_request_thread_events VALUES('w','historical-mr','old-comment',2,'comment','{\"body\":\"handoff done\"}',NULL,'3');")?;
+        conn.execute_batch("CREATE TABLE ticket_implementation_results(collision INTEGER);")?;
+        assert!(migrate_business_notifications_v95_to_v96(conn).is_err());
+        assert!(!table_exists(conn,"merge_request_source_observations")?);
+        assert!(!table_exists(conn,"ticket_implementation_delegations")?);
+        assert_eq!(current_schema_version(conn)?,95);
+        conn.execute_batch("DROP TABLE ticket_implementation_results;")?;
+        migrate_business_notifications_v95_to_v96(conn)?;
+        assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE input_event_id='mr-event:historical-mr:approved-event'",[],|r|r.get::<_,i64>(0))?,2);
+        assert_eq!(conn.query_row("SELECT assignment_id FROM notification_sources WHERE input_event_id='mr-event:historical-mr:approved-event' AND role='worker'",[],|r|r.get::<_,String>(0))?,"", "equal legacy assignment timestamps must not guess ownership");
+        assert_eq!(conn.query_row("SELECT count(*) FROM notification_sources WHERE input_event_id LIKE '%old-comment%'",[],|r|r.get::<_,i64>(0))?,0);
+        assert_eq!(conn.query_row("SELECT orchestrator_assignment_id FROM ticket_implementation_delegations WHERE assignment_id='implementation'",[],|r|r.get::<_,Option<String>>(0))?,None);
+        assert_eq!(conn.query_row("SELECT orchestrator_assignment_id FROM ticket_implementation_delegations WHERE assignment_id='known-implementation'",[],|r|r.get::<_,Option<String>>(0))?,Some("delegation".into()));
+        assert!(conn.execute("UPDATE ticket_implementation_delegations SET orchestrator_assignment_id='delegation'",[]).is_err());
+        assert_eq!(conn.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
+        Ok(())
+    }).unwrap();
+    let inputs = store
+        .list_pending_ticket_assignment_deliveries("w", 100)
+        .unwrap();
+    assert!(
+        inputs
+            .iter()
+            .any(|i| i.input_event_id == "mr-event:historical-mr:approved-event")
+    );
+    for _ in 0..2 {
+        store
+            .observe_merge_request_source("w", "main", "refs/heads/work", "source-b")
+            .unwrap();
+    }
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='mr_source_changed'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                2
+            );
+            Ok(())
+        })
+        .unwrap();
+}
+
+#[tokio::test]
+async fn implementation_result_notification_failure_rolls_back_and_concurrent_retries_share_receipt()
+ {
+    let (dir, store) = tracking_store_fixture().await;
+    tracking_store_assign(&store, "delegation", "assign");
+    let worker = routing_worker_assign(&store, "implementation").worker;
+    let input = implementation_result_input();
+    store.with_conn(|conn| {
+        conn.execute_batch("CREATE TRIGGER reject_result_notification BEFORE INSERT ON notification_sources WHEN NEW.kind='implementation_result' BEGIN SELECT RAISE(ABORT,'injected capture failure'); END;")?;
+        Ok(())
+    }).unwrap();
+    assert!(
+        store
+            .record_implementation_result("w", "tracking", &input, &worker, "tracking-binding")
+            .is_err()
+    );
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                0
+            );
+            conn.execute_batch("DROP TRIGGER reject_result_notification;")?;
+            Ok(())
+        })
+        .unwrap();
+    let barrier = std::sync::Barrier::new(4);
+    let path = dir.path().join("workspace.db");
+    let receipts = std::thread::scope(|scope| {
+        let handles = (0..4)
+            .map(|_| {
+                let worker = &worker;
+                let input = &input;
+                let path = &path;
+                let barrier = &barrier;
+                scope.spawn(move || {
+                    let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
+                    barrier.wait();
+                    store
+                        .record_implementation_result(
+                            "w",
+                            "tracking",
+                            input,
+                            worker,
+                            "tracking-binding",
+                        )
+                        .unwrap()
+                })
+            })
+            .collect::<Vec<_>>();
+        handles
+            .into_iter()
+            .map(|h| h.join().unwrap())
+            .collect::<Vec<_>>()
+    });
+    assert!(receipts.iter().all(|r| r == &receipts[0]));
+    store
+        .with_conn(|conn| {
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM ticket_implementation_results",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT count(*) FROM notification_sources WHERE kind='implementation_result'",
+                    [],
+                    |r| r.get::<_, i64>(0)
+                )?,
+                1
+            );
+            Ok(())
+        })
+        .unwrap();
+}

crates/workspace-server/src/notification_delivery.rs

@@ -9,7 +9,8 @@
 //! | --- | --- | --- |
 //! | Assign, Answer, content and post edits, case, tags, tracking, release | Ticket transaction captures facts; `notification_routing` decides both recipients and non-notification reasons | Saved assignment/Worker requests and ACKs; mutation wakeup and periodic rounds |
 //! | Workspace startup, Orchestrator create/restore/replacement, idle | Wakeup only; never a new request or a second policy | Same saved pending decisions |
-//! | MR/handoff and Backend Job | Separate adapters to be connected by follow-up work | Not inferred from comments or display state |
+//! | MR review/request/revoke/cancel/merge and typed implementation result | Atomic business source capture; central policy | Same durable routing and periodic recovery; never inferred from comments |
+//! | Backend Job | Separately owned adapter | Not inferred from display state |
 //! | Explicit WorkerNotify/WorkerSendInput | Authenticated caller communication | Not automatic business routing |
 //! | Internal SubWorker result and Web subscription | Parent/child execution and UI synchronization respectively | Remain separate; no Ticket policy in Runtime |
 //!

crates/workspace-server/src/notification_routing.rs

@@ -25,6 +25,7 @@ pub enum NotificationEventKind {
     Release,
     AssignmentChanged,
     Result,
+    MergeRequest,
     Unknown,
 }
 impl NotificationEventKind {
@@ -39,6 +40,13 @@ impl NotificationEventKind {
             "tags" | "tags_set" | "tags_changed" => Self::Tags,
             "release" => Self::Release,
             "assignment_changed" => Self::AssignmentChanged,
+            "implementation_result" => Self::Result,
+            "mr_review_requested"
+            | "mr_review"
+            | "mr_review_revoked"
+            | "mr_review_cancelled"
+            | "mr_merge"
+            | "mr_source_changed" => Self::MergeRequest,
             _ => Self::Unknown,
         }
     }
@@ -190,6 +198,7 @@ fn policy(
                 | NotificationEventKind::Answer
                 | NotificationEventKind::ContentEdit
                 | NotificationEventKind::Result
+                | NotificationEventKind::MergeRequest
         ) {
             return ("suppress", R::NotRequest);
         }
@@ -307,6 +316,79 @@ pub(super) fn install(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+/// Capture business facts in the same SQLite transaction as their owners. Historical
+/// MR events are bounded by their original assignment interval, never today's delegation.
+/// Comments are deliberately excluded: neither legacy handoff prose nor review comments
+/// prove a typed result. MR input is an instruction to reread evidence, not approval.
+pub(super) fn install_business_events(conn: &Connection) -> Result<()> {
+    conn.execute_batch(include_str!("implementation_result.sql"))?;
+    let mr = format!("INSERT OR IGNORE INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,actor,event_index,facts_json)
+ SELECT NEW.workspace_id,r.ticket_id,'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id,roles.role,COALESCE(a.assignment_id,''),'mr_' || NEW.kind,NULL,NULL,
+ json_set({FACTS_SQL},'$.merge_request_id',NEW.merge_request_id,'$.event_id',NEW.event_id,'$.subject_ref',COALESCE(json_extract(NEW.payload_json,'$.subject_ref'),json_extract(NEW.payload_json,'$.approved_source_ref')),'$.request_event_id',json_extract(NEW.payload_json,'$.request_event_id'))
+ FROM merge_request_ticket_relations r JOIN typed_tickets t ON t.workspace_id=r.workspace_id AND t.ticket_id=r.ticket_id
+ CROSS JOIN (SELECT 'orchestrator' role UNION ALL SELECT 'worker') roles
+ LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=t.workspace_id AND c.ticket_id=t.ticket_id AND c.role=roles.role
+ LEFT JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id
+ WHERE r.workspace_id=NEW.workspace_id AND r.merge_request_id=NEW.merge_request_id;");
+    conn.execute_batch(&format!("CREATE TRIGGER notification_mr_event AFTER INSERT ON merge_request_thread_events WHEN NEW.kind!='comment' BEGIN {mr} END;"))?;
+    let source = mr.replace("'mr-event:' || NEW.merge_request_id || ':' || NEW.event_id", "'mr-source:' || NEW.merge_request_id || ':' || NEW.sequence")
+        .replace("'mr_' || NEW.kind", "'mr_source_changed'")
+        .replace("NEW.event_id", "NEW.sequence")
+        .replace("COALESCE(json_extract(NEW.payload_json,'$.subject_ref'),json_extract(NEW.payload_json,'$.approved_source_ref'))", "NEW.subject_ref")
+        .replace("json_extract(NEW.payload_json,'$.request_event_id')", "NULL");
+    conn.execute_batch(&format!("CREATE TRIGGER notification_mr_source AFTER INSERT ON merge_request_source_observations WHEN NEW.previous_subject_ref IS NOT NULL BEGIN {source} END;"))?;
+    // Recover retained review/merge evidence once. A source identity is the immutable
+    // MR event, so re-review is distinct and retry cannot duplicate an input.
+    // Equal/ambiguous assignment timestamps cannot prove event-time ownership;
+    // retain a no-recipient source instead of guessing by row order.
+    let historical = mr.replace("NEW.", "e.").replace(
+        "FROM merge_request_ticket_relations r", "FROM merge_request_thread_events e JOIN merge_request_ticket_relations r ON r.workspace_id=e.workspace_id AND r.merge_request_id=e.merge_request_id"
+    ).replace(
+        "LEFT JOIN ticket_current_worker_assignments c ON c.workspace_id=t.workspace_id AND c.ticket_id=t.ticket_id AND c.role=roles.role\n LEFT JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.assignment_id=c.assignment_id",
+        "LEFT JOIN ticket_worker_assignments a ON a.workspace_id=t.workspace_id AND a.ticket_id=t.ticket_id AND a.role=roles.role AND a.assignment_id=(SELECT CASE WHEN COUNT(*)=1 THEN MIN(h.assignment_id) END FROM ticket_worker_assignments h WHERE h.workspace_id=t.workspace_id AND h.ticket_id=t.ticket_id AND h.role=roles.role AND h.assigned_at=(SELECT MAX(prior.assigned_at) FROM ticket_worker_assignments prior WHERE prior.workspace_id=t.workspace_id AND prior.ticket_id=t.ticket_id AND prior.role=roles.role AND prior.assigned_at<e.created_at))"
+    ).replace("WHERE r.workspace_id=e.workspace_id AND r.merge_request_id=e.merge_request_id;", "WHERE e.kind!='comment';");
+    conn.execute_batch(&historical)?;
+    conn.execute_batch(&format!("CREATE TRIGGER notification_implementation_result AFTER INSERT ON ticket_implementation_results BEGIN
+ INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,actor,event_index,facts_json)
+ SELECT NEW.workspace_id,NEW.ticket_id,'implementation-result:' || NEW.result_id,'orchestrator',a.assignment_id,'implementation_result','worker:' || NEW.runtime_id || '/' || NEW.worker_id,NULL,
+ json_set({FACTS_SQL},'$.result_id',NEW.result_id,'$.implementation_assignment_id',NEW.assignment_id)
+ FROM ticket_worker_assignments a JOIN typed_tickets t ON t.workspace_id=a.workspace_id AND t.ticket_id=a.ticket_id
+ WHERE a.workspace_id=NEW.workspace_id AND a.assignment_id=NEW.orchestrator_assignment_id;
+ END;"))?;
+    Ok(())
+}
+
+impl SqliteWorkspaceStore {
+    /// Record a provider-resolved source observation, not approval. Git has no local
+    /// transactional event log: transitions are captured when the existing MR paths
+    /// observe them. Identical observations (including concurrent scans) are deduplicated.
+    pub(crate) fn observe_merge_request_source(
+        &self,
+        workspace: &str,
+        repository: &str,
+        selector: &str,
+        subject: &str,
+    ) -> Result<()> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let ids = tx.prepare("SELECT merge_request_id FROM merge_requests WHERE workspace_id=?1 AND repository_id=?2 AND selector_from=?3 AND state='open'")?.query_map(params![workspace,repository,selector],|r|r.get::<_,String>(0))?.collect::<rusqlite::Result<Vec<_>>>()?;
+            for id in ids {
+                let previous: Option<String> = tx.query_row("SELECT subject_ref FROM merge_request_source_observations WHERE workspace_id=?1 AND merge_request_id=?2 ORDER BY sequence DESC LIMIT 1", params![workspace,id], |r|r.get(0)).optional()?;
+                // New MRs may already have a review before the first observation.
+                let previous = match previous {
+                    Some(previous) => Some(previous),
+                    None => tx.query_row("SELECT json_extract(payload_json,'$.subject_ref') FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2 AND json_extract(payload_json,'$.subject_ref') IS NOT NULL ORDER BY sequence DESC LIMIT 1",params![workspace,id],|r|r.get(0)).optional()?,
+                };
+                if previous.as_deref() != Some(subject) {
+                    tx.execute("INSERT INTO merge_request_source_observations(workspace_id,merge_request_id,subject_ref,previous_subject_ref,observed_at) VALUES(?1,?2,?3,?4,?5)",params![workspace,id,subject,previous,chrono::Utc::now().to_rfc3339()])?;
+                }
+            }
+            tx.commit()?;
+            Ok(())
+        })
+    }
+}
+
 pub(super) fn verify(conn: &Connection) -> Result<()> {
     let mut names = vec![
         "notification_sources".to_owned(),
@@ -339,6 +421,27 @@ pub(super) fn verify(conn: &Connection) -> Result<()> {
     ] {
         names.push(name.into());
     }
+    if current_schema_version(conn)? >= 96 {
+        names.extend(
+            [
+                "ticket_implementation_delegations",
+                "ticket_implementation_results",
+                "implementation_delegation_capture",
+                "implementation_results_no_update",
+                "implementation_results_no_delete",
+                "implementation_delegations_no_update",
+                "implementation_delegations_no_delete",
+                "notification_mr_event",
+                "notification_implementation_result",
+                "notification_mr_source",
+                "merge_request_source_observations",
+                "mr_source_observations_no_update",
+                "mr_source_observations_no_delete",
+            ]
+            .into_iter()
+            .map(str::to_owned),
+        );
+    }
     for name in names {
         let present: bool = conn.query_row(
             "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name=?1)",

crates/workspace-server/src/notification_routing_tests.rs

@@ -843,6 +843,7 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() {
     migrate_assignment_delivery_v92_to_v93(&conn).unwrap();
     migrate_ticket_posts_v93_to_v94(&conn).unwrap();
     migrate_notification_routing_v94_to_v95(&conn).unwrap();
+    migrate_business_notifications_v95_to_v96(&conn).unwrap();
     let snapshot = |conn: &Connection| -> Vec<(String, String)> {
         conn.prepare("SELECT name,sql FROM sqlite_schema WHERE name LIKE 'notification_%' OR name LIKE 'ticket_assignment_content%' OR name IN ('ticket_assignment_title_input','ticket_assignment_initial_post_input') ORDER BY name").unwrap().query_map([],|r|Ok((r.get(0)?,r.get(1)?))).unwrap().collect::<rusqlite::Result<_>>().unwrap()
     };

crates/workspace-server/src/server.rs

@@ -3356,6 +3356,10 @@ fn canonical_ticket_operation_workspace_id(path: &str) -> Option<&str> {
         && resource == "tickets"
         && !ticket_id.is_empty()
         && match operation {
+            "implementation-results" => match segments.next() {
+                None => true,
+                Some(id) => !id.is_empty() && segments.next().is_none(),
+            },
             "case" | "assignment-diagnostics" | "assignment-answer" => segments.next().is_none(),
             "assignment-tracking" => match segments.next() {
                 None => true,
@@ -6934,6 +6938,11 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
             service.clone(),
         ))
         .merge(server_api::server_api_axum::ticket_assignment_answer_browser(service.clone()))
+        .merge(server_api::server_api_axum::ticket_implementation_result(
+            service.clone(),
+        ))
+        .merge(server_api::server_api_axum::ticket_implementation_result_browser(service.clone()))
+        .merge(server_api::server_api_axum::ticket_implementation_result_get(service.clone()))
         .merge(server_api::server_api_axum::ticket_assignment_tracking_get(
             service.clone(),
         ))
@@ -11221,6 +11230,99 @@ impl server_api::ServerApi for ServerApiContractService {
             .await
     }
 
+    async fn ticket_implementation_result(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        request: server_api::TicketImplementationResultRequest,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        let api = self.workspace_api()?.clone();
+        api.host_executors
+            .host()?
+            .run(move || {
+                let result = (|| -> ApiResult<_> {
+                    validate_workspace_scope(&api, &workspace_id)?;
+                    let ticket =
+                        resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
+                    let source = context.runtime_source.as_ref().ok_or_else(|| {
+                        Error::WorkspacePermissionDenied(
+                            "Implementation results require an authenticated Runtime Worker".into(),
+                        )
+                    })?;
+                    let worker = RuntimeWorkerRef::new(
+                        &source.runtime_id,
+                        source.worker_id.as_ref().ok_or_else(|| {
+                            Error::WorkspacePermissionDenied(
+                                "Implementation results require Worker identity".into(),
+                            )
+                        })?,
+                    );
+                    let receipt = api.config_store.record_implementation_result(
+                        &workspace_id,
+                        &ticket,
+                        &request,
+                        &worker,
+                        &source.authentication_binding_id,
+                    )?;
+                    recover_ticket_notifications(&api);
+                    Ok(receipt)
+                })();
+                result
+                    .map_err(ApiError::from)
+                    .map_err(ApiError::into_repository_api_error)
+            })
+            .await
+    }
+
+    async fn ticket_implementation_result_browser(
+        &self,
+        context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        request: server_api::TicketImplementationResultRequest,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        self.ticket_implementation_result(context, workspace_id, ticket_id, request)
+            .await
+    }
+
+    async fn ticket_implementation_result_get(
+        &self,
+        _context: server_api::ServerRequestContext,
+        workspace_id: String,
+        ticket_id: String,
+        result_id: String,
+    ) -> std::result::Result<
+        server_api::TicketImplementationResultReceipt,
+        server_api::RepositoryApiError,
+    > {
+        let api = self.workspace_api()?.clone();
+        api.host_executors
+            .host()?
+            .run(move || {
+                let result = (|| -> ApiResult<_> {
+                    validate_workspace_scope(&api, &workspace_id)?;
+                    let ticket =
+                        resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
+                    Ok(api.config_store.implementation_result(
+                        &workspace_id,
+                        &ticket,
+                        &result_id,
+                    )?)
+                })();
+                result
+                    .map_err(ApiError::from)
+                    .map_err(ApiError::into_repository_api_error)
+            })
+            .await
+    }
+
     async fn ticket_assignment_tracking_get(
         &self,
         _context: server_api::ServerRequestContext,
@@ -16727,8 +16829,16 @@ fn observe_published_source_ref(
     repository_id: &str,
     selector: &str,
 ) -> ApiResult<RepositoryRefObservation> {
-    observe_published_merge_ref(api, workspace_id, runtime_id, repository_id, selector)
-        .map_err(remap_source_ref_error)
+    let observation =
+        observe_published_merge_ref(api, workspace_id, runtime_id, repository_id, selector)
+            .map_err(remap_source_ref_error)?;
+    api.config_store.observe_merge_request_source(
+        workspace_id,
+        repository_id,
+        selector,
+        &observation.resolved_ref,
+    )?;
+    Ok(observation)
 }
 
 /// Before opening an MR or granting review of an open MR, re-read the current

crates/workspace-server/src/server_assignment_delivery_tests.rs

@@ -5,6 +5,7 @@
 use super::*;
 use session_store::{LogEntry, Store, WorkerSessionStore};
 use worker_runtime::execution as delivery_execution;
+include!("server_implementation_result_tests.rs");
 
 fn delivery_assignment_path(ticket: &str) -> String {
     format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{ticket}/assignments/orchestrator")

crates/workspace-server/src/server_implementation_result_tests.rs

@@ -0,0 +1,325 @@
+// Included in the production persistent-Worker delivery fixture. Only the LLM is
+// scripted. Signed Runtime HTTP, assignment fencing, SQLite and session receipts are real.
+async fn result_signed_http(
+    api: &WorkspaceApi,
+    identity: &RuntimeIdentityMaterial,
+    worker: &RuntimeWorkerRef,
+    ticket: &str,
+    body: &Value,
+    expected: StatusCode,
+) -> Value {
+    let path = format!("/workspaces/{TEST_WORKSPACE_ID}/tickets/{ticket}/implementation-results");
+    let response = build_router(api.clone())
+        .oneshot(runtime_source_request(
+            identity,
+            Some(&worker.worker_id),
+            "POST",
+            &path,
+            serde_json::to_vec(body).unwrap(),
+        ))
+        .await
+        .unwrap();
+    let status = response.status();
+    let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+    assert_eq!(status, expected, "{}", String::from_utf8_lossy(&bytes));
+    serde_json::from_slice(&bytes).unwrap()
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn implementation_result_signed_http_reaches_persistent_orchestrator_once_without_mr() {
+    let dir = tempfile::tempdir().unwrap();
+    let (mut api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let identity = RuntimeIdentityMaterial::generate(EMBEDDED_WORKER_RUNTIME_ID).unwrap();
+    configure_runtime_request_auth(&mut api, &identity, EMBEDDED_WORKER_RUNTIME_ID);
+    let ticket = delivery_ticket(&api, "Return repository-free research");
+    let assigned = delivery_assign(&api, &ticket.id, "research-delegation").await;
+    let delegation = assigned["assignment"]["assignment_id"].as_str().unwrap();
+    let orchestrator = delivery_worker(start_attention_test_orchestrator(&api).await);
+    seed_worker_source_member(&api, &orchestrator.runtime_id, &orchestrator.worker_id);
+    delivery_wait_idle(&api, &orchestrator, 1).await;
+    let initial = delivery_tracking(&api, &ticket.id, delegation);
+    delivery_self_report(
+        &api,
+        &identity,
+        &orchestrator,
+        &initial,
+        "inprogress",
+        "research-progress",
+    )
+    .await;
+    let coder = spawn_ticket_check_source(&api, "research-coder");
+    seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id);
+    delivery_assign_worker(&api, &ticket.id, &coder, "research-implementation", None);
+    recover_ticket_notifications(&api);
+    let before = delivery_tracking(&api, &ticket.id, delegation);
+    let input = json!({"operation_key":"research-result","assignment_id":"research-implementation","orchestrator_assignment_id":delegation,"kind":"result","body":"Documented findings; no repository or MR was required.","merge_request":null});
+    // The registered Orchestrator is not the implementation assignee and cannot forge this result.
+    result_signed_http(
+        &api,
+        &identity,
+        &orchestrator,
+        &ticket.id,
+        &input,
+        StatusCode::FORBIDDEN,
+    )
+    .await;
+    let receipt =
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    delivery_wait_idle(&api, &orchestrator, 2).await;
+    assert_eq!(
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await,
+        receipt
+    );
+    let mut changed = input.clone();
+    changed["body"] = json!("Different result");
+    result_signed_http(
+        &api,
+        &identity,
+        &coder,
+        &ticket.id,
+        &changed,
+        StatusCode::CONFLICT,
+    )
+    .await;
+    let mut forged = input.clone();
+    forged["operation_key"] = json!("forged-result");
+    forged["merge_request"] = json!({"merge_request_id":"missing","subject_ref":"fake"});
+    result_signed_http(
+        &api,
+        &identity,
+        &coder,
+        &ticket.id,
+        &forged,
+        StatusCode::BAD_REQUEST,
+    )
+    .await;
+    let id = receipt["result_id"].as_str().unwrap();
+    assert!(
+        delivery_commits(&api, &orchestrator)[1].contains(&format!("implementation-result:{id}"))
+    );
+    assert_eq!(delivery_tracking(&api, &ticket.id, delegation), before);
+    assert_eq!(
+        api.authority.ticket(&ticket.id).unwrap().state,
+        server_api::BrowserTicketState::Open
+    );
+    let saved = delivery_http(
+        &api,
+        "GET",
+        &format!(
+            "/workspaces/{TEST_WORKSPACE_ID}/tickets/{}/implementation-results/{id}",
+            ticket.id
+        ),
+        None,
+        StatusCode::OK,
+    )
+    .await;
+    assert_eq!(saved, receipt);
+    let attempts = execution.attempts.lock().unwrap().len();
+    for _ in 0..3 {
+        recover_ticket_notifications(&api);
+    }
+    assert_eq!(execution.attempts.lock().unwrap().len(), attempts);
+    assert_eq!(delivery_receipts(&api, &orchestrator).len(), 2);
+    delivery_stop(&api, &coder);
+    delivery_stop(&api, &orchestrator);
+}
+
+// MR event creation uses the production store and registered reviewer capability.
+// Provider observations/child registration are fixture boundaries here; the existing
+// authenticated MR router suite owns Git publication and trusted child attestation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn mr_result_recovery_and_handoff_are_distinct_persistent_inputs() {
+    let dir = tempfile::tempdir().unwrap();
+    let (mut api, execution, _) = delivery_real_fixture(dir.path(), false).await;
+    let identity = RuntimeIdentityMaterial::generate(EMBEDDED_WORKER_RUNTIME_ID).unwrap();
+    configure_runtime_request_auth(&mut api, &identity, EMBEDDED_WORKER_RUNTIME_ID);
+    let ticket = delivery_ticket(&api, "MR result followed by implementation handoff");
+    let assigned = delivery_assign(&api, &ticket.id, "mr-delegation").await;
+    let delegation = assigned["assignment"]["assignment_id"].as_str().unwrap();
+    let orchestrator = delivery_worker(start_attention_test_orchestrator(&api).await);
+    seed_worker_source_member(&api, &orchestrator.runtime_id, &orchestrator.worker_id);
+    delivery_wait_idle(&api, &orchestrator, 1).await;
+    let initial = delivery_tracking(&api, &ticket.id, delegation);
+    delivery_self_report(
+        &api,
+        &identity,
+        &orchestrator,
+        &initial,
+        "inprogress",
+        "mr-progress",
+    )
+    .await;
+    let coder = spawn_ticket_check_source(&api, "mr-coder");
+    seed_worker_source_member(&api, &coder.runtime_id, &coder.worker_id);
+    delivery_assign_worker(&api, &ticket.id, &coder, "mr-implementation", None);
+    seed_test_repository(&api, TEST_REPOSITORY_ID);
+    api.config.repositories = test_server_config(dir.path()).repositories;
+    let store = merge_request_store(&api, TEST_WORKSPACE_ID).unwrap();
+    let auth = merge_request::MergeRequestAuth {
+        workspace_id: TEST_WORKSPACE_ID.into(),
+        repository_id: TEST_REPOSITORY_ID.into(),
+        runtime_id: coder.runtime_id.clone(),
+        worker_id: coder.worker_id.clone(),
+        assignment_id: "mr-implementation".into(),
+    };
+    store
+        .open_merge_request(merge_request::OpenMergeRequest {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            repository_id: TEST_REPOSITORY_ID.into(),
+            selector_from: "refs/heads/work".into(),
+            selector_to: "refs/heads/develop".into(),
+            summary: String::new(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    store
+        .register_reviewer_child_session(merge_request::RegisterReviewerChildSession {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            parent_runtime_id: coder.runtime_id.clone(),
+            parent_worker_id: coder.worker_id.clone(),
+            child_session_id: "result-reviewer".into(),
+            reviewer_profile: "builtin:reviewer".into(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    let digest = api.authority.ticket(&ticket.id).unwrap().content_digest;
+    let requested = store
+        .request_review(merge_request::RequestMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            ticket_content_digest: digest,
+            ticket_merge_request_subjects: vec![merge_request::MergeRequestReviewSubject {
+                merge_request_id: "result-mr".into(),
+                subject_ref: "source-a".into(),
+            }],
+            subject_ref: "source-a".into(),
+            child_session_id: "result-reviewer".into(),
+            capability_token: "result-review-capability".into(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    let approved = store
+        .submit_review(merge_request::SubmitMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            current_subject_ref: "source-a".into(),
+            capability_token: "result-review-capability".into(),
+            decision: merge_request::ReviewDecision::Approve,
+            body: "Approved exact source".into(),
+            findings: vec![],
+            now: Utc::now(),
+        })
+        .unwrap();
+    // No mutation callback: both saved MR events must be found by normal recovery.
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 3).await;
+    let input = json!({"operation_key":"approved-handoff","assignment_id":"mr-implementation","orchestrator_assignment_id":delegation,"kind":"handoff","body":"Implementation and validation complete; integration remains separate.","merge_request":{"merge_request_id":"result-mr","subject_ref":"source-a"}});
+    let receipt =
+        result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    delivery_wait_idle(&api, &orchestrator, 4).await;
+    let commits = delivery_commits(&api, &orchestrator);
+    assert!(
+        commits
+            .iter()
+            .any(|s| s.contains(&requested.request_event.event_id))
+    );
+    assert!(commits.iter().any(|s| s.contains(&approved.event_id)));
+    assert!(
+        commits
+            .iter()
+            .any(|s| s.contains(receipt["result_id"].as_str().unwrap()))
+    );
+    store
+        .revoke_review(merge_request::RevokeMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            review_event_id: approved.event_id,
+            reason: "Approval withdrawn".into(),
+            auth: auth.clone(),
+            now: Utc::now(),
+        })
+        .unwrap();
+    api.config_store
+        .observe_merge_request_source(
+            TEST_WORKSPACE_ID,
+            TEST_REPOSITORY_ID,
+            "refs/heads/work",
+            "source-b",
+        )
+        .unwrap();
+    api.config_store
+        .observe_merge_request_source(
+            TEST_WORKSPACE_ID,
+            TEST_REPOSITORY_ID,
+            "refs/heads/work",
+            "source-b",
+        )
+        .unwrap();
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 6).await;
+    for (child, requested_source, observed_source) in [
+        ("cancelled-reviewer", "source-b", "source-c"),
+        ("changes-reviewer", "source-c", "source-c"),
+    ] {
+        store
+            .register_reviewer_child_session(merge_request::RegisterReviewerChildSession {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                parent_runtime_id: coder.runtime_id.clone(),
+                parent_worker_id: coder.worker_id.clone(),
+                child_session_id: child.into(),
+                reviewer_profile: "builtin:reviewer".into(),
+                now: Utc::now(),
+            })
+            .unwrap();
+        store
+            .request_review(merge_request::RequestMergeRequestReview {
+                merge_request_id: "result-mr".into(),
+                ticket_id: ticket.id.clone(),
+                ticket_content_digest: api.authority.ticket(&ticket.id).unwrap().content_digest,
+                ticket_merge_request_subjects: vec![merge_request::MergeRequestReviewSubject {
+                    merge_request_id: "result-mr".into(),
+                    subject_ref: requested_source.into(),
+                }],
+                subject_ref: requested_source.into(),
+                child_session_id: child.into(),
+                capability_token: child.into(),
+                auth: auth.clone(),
+                now: Utc::now(),
+            })
+            .unwrap();
+        let submitted = store.submit_review(merge_request::SubmitMergeRequestReview {
+            merge_request_id: "result-mr".into(),
+            ticket_id: ticket.id.clone(),
+            current_subject_ref: observed_source.into(),
+            capability_token: child.into(),
+            decision: merge_request::ReviewDecision::RequestChanges,
+            body: "Recheck implementation".into(),
+            findings: vec![],
+            now: Utc::now(),
+        });
+        assert_eq!(submitted.is_err(), requested_source != observed_source);
+    }
+    recover_ticket_notifications(&api);
+    delivery_wait_idle(&api, &orchestrator, 10).await;
+    let attempts = execution.attempts.lock().unwrap().len();
+    result_signed_http(&api, &identity, &coder, &ticket.id, &input, StatusCode::OK).await;
+    recover_ticket_notifications(&api);
+    assert_eq!(execution.attempts.lock().unwrap().len(), attempts);
+    assert_eq!(delivery_receipts(&api, &orchestrator).len(), 10);
+    assert_eq!(
+        api.authority.ticket(&ticket.id).unwrap().state,
+        server_api::BrowserTicketState::Open
+    );
+    assert_eq!(
+        delivery_tracking(&api, &ticket.id, delegation)
+            .tracking
+            .unwrap()
+            .status,
+        ticket::TicketAssignmentTrackingStatus::Inprogress
+    );
+    delivery_stop(&api, &coder);
+    delivery_stop(&api, &orchestrator);
+}

crates/workspace-server/src/store.rs

@@ -70,6 +70,8 @@ pub use notification_routing::{
 };
 #[path = "assignment_tracking.rs"]
 mod assignment_tracking;
+#[path = "implementation_result.rs"]
+mod implementation_result;
 pub use assignment_delivery::TicketAssignmentDelivery;
 pub use assignment_tracking::TicketOrchestratorAssignmentSourceContext;
 pub use ticket::{
@@ -79,7 +81,7 @@ pub use ticket::{
 };
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 95;
+const LATEST_SCHEMA_VERSION: i64 = 96;
 const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
 const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
 const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
@@ -362,6 +364,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "Common notification routing and durable Worker delivery",
         apply: migrate_notification_routing_v94_to_v95,
     },
+    Migration {
+        version: 96,
+        name: "MR notifications and authenticated implementation results",
+        apply: migrate_business_notifications_v95_to_v96,
+    },
 ];
 
 #[derive(Clone, Copy)]
@@ -14123,11 +14130,30 @@ fn migrate_notification_routing_v94_to_v95(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+fn migrate_business_notifications_v95_to_v96(conn: &Connection) -> Result<()> {
+    if current_schema_version(conn)? != 95 {
+        return Err(Error::Store(
+            "Business notifications migration requires schema 95".into(),
+        ));
+    }
+    // Older Workspace databases may not have opened the MR domain yet. Complete
+    // its owner-managed, idempotent migration before composing business triggers.
+    // A failed host migration leaves that independent domain upgrade intact.
+    merge_request::migrate(conn)?;
+    let tx = conn.unchecked_transaction()?;
+    notification_routing::install_business_events(&tx)?;
+    tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(96,'MR notifications and authenticated implementation results')", [])?;
+    verify_workspace_resource_constraints(&tx)?;
+    tx.commit()?;
+    Ok(())
+}
+
 fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
     create_workspace_schema_structure(conn)?;
     assignment_delivery::install(conn)?;
     assignment_delivery::install_post_inputs(conn)?;
     notification_routing::install(conn)?;
+    notification_routing::install_business_events(conn)?;
     ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
         Error::Store(format!(
             "fresh Workspace Ticket schema verification failed: {error}"
@@ -15141,6 +15167,7 @@ mod tests {
     include!("assignment_tracking_tests.rs");
     include!("assignment_delivery_tests.rs");
     include!("notification_routing_tests.rs");
+    include!("implementation_result_tests.rs");
     include!("store_workspace_config_tests.rs");
 
     fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
@@ -17354,6 +17381,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                 WorkspaceSchemaMigrationStep {version:93, name:"Durable assignment deliveries and source-bound startup fences".into()},
                 WorkspaceSchemaMigrationStep {version:94, name:"Canonical Ticket posts".into()},
                 WorkspaceSchemaMigrationStep {version:95, name:"Common notification routing and durable Worker delivery".into()},
+                WorkspaceSchemaMigrationStep {version:96, name:"MR notifications and authenticated implementation results".into()},
             ]
         );
 
@@ -17459,6 +17487,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (93, "Durable assignment deliveries and source-bound startup fences".to_string()),
 (94, "Canonical Ticket posts".to_string()),
 (95, "Common notification routing and durable Worker delivery".to_string()),
+(96, "MR notifications and authenticated implementation results".to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -20078,7 +20107,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                          (92, "Ticket-owned assignment tracking".into()),
                          (93, "Durable assignment deliveries and source-bound startup fences".into()),
                          (94, "Canonical Ticket posts".into()),
-                         (95, "Common notification routing and durable Worker delivery".into())]);
+                         (95, "Common notification routing and durable Worker delivery".into()),
+                         (96, "MR notifications and authenticated implementation results".into())]);
                 for (table, rows) in unchanged.iter().zip(before.iter()) {
                     assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                 }

crates/workspace-server/src/workspace_deletion.rs

@@ -26,6 +26,9 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "flow_sources",
     "memory_staging_records",
     "memory_staging_resolutions",
+    "merge_request_source_observations",
+    "ticket_implementation_results",
+    "ticket_implementation_delegations",
     "merge_request_review_grants",
     "merge_request_reviewer_child_sessions",
     "merge_request_thread_events",
@@ -490,6 +493,9 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
             let mut ticket_delete_guards = Vec::new();
             for (table, trigger) in [
                 ("notification_assignment_targets", "notification_assignment_targets_no_delete"),
+                ("merge_request_source_observations", "mr_source_observations_no_delete"),
+                ("ticket_implementation_results", "implementation_results_no_delete"),
+                ("ticket_implementation_delegations", "implementation_delegations_no_delete"),
                 ("notification_sources", "notification_sources_no_delete"),
                 ("notification_decisions", "notification_decisions_no_delete"),
                 ("notification_requests", "notification_requests_no_delete"),