refactor(work): run adjudications and placements as writer jobs

crates/tracedecay-application/src/work/registered.rs

@@ -283,6 +283,142 @@ impl RegisteredWorkApplicationServicesV1 {
             .unwrap_or_else(|_| Err(run_control_unavailable()))
     }
 
+    /// Records one duplicate adjudication as a single writer transaction: the
+    /// replay check, the revision compare and the append commit together.
+    /// `current_evidence` is read before the job, outside the writer.
+    pub async fn adjudicate_duplicate(
+        &self,
+        context: tracedecay_contracts::RequestContext,
+        command: tracedecay_domain::WorkDuplicateAdjudicationCommandV1,
+        current_evidence: tracedecay_domain::WorkDuplicateAdjudicationEvidenceV1,
+    ) -> Result<
+        tracedecay_contracts::WorkDuplicateAdjudicationAppendOutcomeV1,
+        tracedecay_contracts::ApplicationProblem,
+    > {
+        self.storage
+            .transact(move |storage| {
+                tracedecay_contracts::WorkDuplicateAdjudicationServiceV1::new(storage).adjudicate(
+                    &context,
+                    command.clone(),
+                    current_evidence.clone(),
+                )
+            })
+            .await
+            .unwrap_or_else(|_| {
+                Err(
+                    tracedecay_contracts::WorkDuplicateAdjudicationStorageErrorV1::Unavailable
+                        .into(),
+                )
+            })
+    }
+
+    /// Records one leak adjudication as a single writer transaction: the
+    /// command replay check, the verdict over `evidence` and the
+    /// revision-compared append commit together. `evidence` is collected
+    /// before the job, outside the writer.
+    pub async fn adjudicate_leak<E>(
+        &self,
+        context: tracedecay_contracts::RequestContext,
+        command: tracedecay_contracts::AdjudicateWorkLeakCommandV1,
+        evidence: E,
+        scan_started_at: tracedecay_domain::UtcMicros,
+        scan_deadline: tracedecay_domain::UtcMicros,
+    ) -> Result<
+        tracedecay_contracts::WorkLeakAdjudicationOutcomeV1,
+        tracedecay_contracts::ApplicationProblem,
+    >
+    where
+        E: tracedecay_contracts::WorkLeakEvidencePortV1 + Clone + 'static,
+    {
+        self.storage
+            .transact(move |storage| {
+                tracedecay_contracts::WorkLeakAdjudicationServiceV1::new(storage, evidence.clone())
+                    .adjudicate(&context, command.clone(), scan_started_at, scan_deadline)
+            })
+            .await
+            .unwrap_or_else(|_| {
+                Err(tracedecay_contracts::WorkLeakAdjudicationStorageErrorV1::Unavailable.into())
+            })
+    }
+
+    /// Admits one placement as a single writer transaction: the replay check,
+    /// the exclusivity read and the publication commit together. `observe`
+    /// reads the target (Git) before the job, never on the writer.
+    pub async fn admit_placement(
+        &self,
+        context: tracedecay_contracts::RequestContext,
+        command: tracedecay_contracts::AdmitWorkPlacementCommand,
+        observe: impl FnOnce(
+            &tracedecay_domain::WorkPlacementTargetV1,
+        ) -> Result<
+            tracedecay_domain::WorkPlacementObservationV1,
+            tracedecay_contracts::ApplicationProblem,
+        >,
+    ) -> Result<tracedecay_domain::WorkPlacementV1, tracedecay_contracts::ApplicationProblem> {
+        let observation = observe(&command.target)?;
+        self.storage
+            .transact(move |storage| {
+                tracedecay_contracts::WorkPlacementService::new(storage).admit_placement(
+                    &context,
+                    command.clone(),
+                    |_| Ok(observation),
+                )
+            })
+            .await
+            .unwrap_or_else(|_| Err(placement_unavailable()))
+    }
+
+    /// Releases one placement. The placement is read on a reader and its
+    /// target observed before the job; the job then publishes only while the
+    /// placement still has the observed target and the expected authority
+    /// version, so the Git observation never runs on the writer.
+    pub async fn release_placement(
+        &self,
+        context: tracedecay_contracts::RequestContext,
+        command: tracedecay_contracts::ReleaseWorkPlacementCommand,
+        observe: impl FnOnce(
+            &tracedecay_domain::WorkPlacementTargetV1,
+        ) -> Result<
+            tracedecay_domain::WorkPlacementObservationV1,
+            tracedecay_contracts::ApplicationProblem,
+        >,
+    ) -> Result<tracedecay_domain::WorkPlacementV1, tracedecay_contracts::ApplicationProblem> {
+        let authority = tracedecay_contracts::work_authority(&context)?;
+        let identity = tracedecay_domain::WorkPlacementIdentityV1::new(
+            command.task_id.clone(),
+            command.run_id.clone(),
+        );
+        let current = self
+            .storage
+            .load_placement_async(&authority, &identity)
+            .await?
+            .ok_or(tracedecay_contracts::WorkPlacementStorageError::NotFoundOrNotAuthorized)?;
+        if current.authority_version() != command.expected_authority_version {
+            return Err(tracedecay_contracts::WorkPlacementStorageError::AuthorityConflict.into());
+        }
+        let observation = observe(current.target())?;
+        let observed_target = current.target().clone();
+        self.storage
+            .transact(move |storage| {
+                tracedecay_contracts::WorkPlacementService::new(storage).release(
+                    &context,
+                    command.clone(),
+                    |target| {
+                        if target == &observed_target {
+                            Ok(observation)
+                        } else {
+                            Err(
+                                tracedecay_contracts::WorkPlacementStorageError::AuthorityConflict
+                                    .into(),
+                            )
+                        }
+                    },
+                )
+            })
+            .await
+            .unwrap_or_else(|_| Err(placement_unavailable()))
+    }
+
     pub fn attempts(
         &self,
     ) -> &tracedecay_contracts::WorkAttemptService<
@@ -422,6 +558,10 @@ pub fn authorized_work_intelligence_service(
     ))
 }
 
+fn placement_unavailable() -> tracedecay_contracts::ApplicationProblem {
+    tracedecay_contracts::WorkPlacementStorageError::Unavailable.into()
+}
+
 fn run_control_unavailable() -> tracedecay_contracts::ApplicationProblem {
     tracedecay_contracts::WorkRunControlStorageError::Unavailable.into()
 }

crates/tracedecay-application/tests/application_suite/main.rs

@@ -17,6 +17,7 @@ mod observability_rollup_convergence;
 // root: the work_rollup bench mounts it, and the suite itself uses a subset.
 pub mod observability_runtime_contract;
 mod registered_scope_route;
+mod work_leaf_jobs;
 mod work_run_control_jobs;
 mod work_service_composition;
 mod workflow_topology_contract;