Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: project Job receipts and pinned notification recipients

Browse Source
This commit is contained in:
Hare
2026-10-15 18:28:11 +09:00
parent c155259c8e
commit b9e7d5675b
11 changed files with 59 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server-api/src/notification.rs
+2
View File
@@ -61,6 +61,8 @@ pub struct NotificationDiagnostic {
pub purpose: Option<String>,
pub delivery_id: Option<String>,
pub stage: NotificationDiagnosticStage,
/// A retained pre-cutover Job receipt contributes to this projection.
pub legacy_evidence: bool,
/// Conservative: a persisted request without ACK is enough; this does not prove a send occurred.
/// Can remain true alongside a later preflight failure, rejection or historical recipient.
pub acceptance_outcome_unknown: bool,
crates/worker/src/feature/builtin/ticket/notification_diagnostics/tests.rs
+1 -1
View File
@@ -79,7 +79,7 @@ async fn notification_tool_and_wip_share_bounded_read_contract() {
let client = Arc::new(Client {
requests: Mutex::new(vec![]),
status: 200,
body: json!({"items":[{"source_id":"notification-source:5","source_event_id":"mr-event:m:e","event_kind":"mr_review","recipient_role":"orchestrator","stage":"acceptance_unknown","acceptance_outcome_unknown":true,"target_resolution":"multiple_authorized_targets","retry_candidate":true,"model_processing_confirmed":false}],"next_cursor":null,"source_checks":[]}).to_string(),
body: json!({"items":[{"source_id":"notification-source:5","source_event_id":"mr-event:m:e","event_kind":"mr_review","recipient_role":"orchestrator","stage":"acceptance_unknown","legacy_evidence":false,"acceptance_outcome_unknown":true,"target_resolution":"multiple_authorized_targets","retry_candidate":true,"model_processing_confirmed":false}],"next_cursor":null,"source_checks":[]}).to_string(),
});
let (meta, tool) = definition(client.clone())();
let schema = jsonschema::validator_for(&meta.input_schema).unwrap();
crates/workspace-server/src/job_notification_tests.rs
+28
View File
@@ -142,6 +142,21 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur
store.route_job_notification("w", &source, &facts).unwrap(),
disposition == "deliver"
);
let diagnostics=store.notification_diagnostics("w",&server_api::NotificationDiagnosticsQuery{job_id:Some(name.into()),..Default::default()}).unwrap();
let item=&diagnostics.items[0];
assert_eq!(item.disposition.as_deref(),Some(disposition));
assert_eq!(item.reason.as_deref(),Some(reason));
assert_eq!(item.job_id.as_deref(),Some(name));
assert!(diagnostics.source_checks[0].captured);
assert_eq!(diagnostics.source_checks[0].origin_present,Some(true));
assert!(!item.model_processing_confirmed && !item.legacy_evidence);
if disposition=="deliver" {
assert_eq!(item.stage,server_api::NotificationDiagnosticStage::RequestPersisted);
assert_eq!(item.target_worker_id.as_deref(),Some("replacement"));
} else {
assert!(!item.retry_candidate);
assert_eq!(item.stage,if disposition=="suppress"{server_api::NotificationDiagnosticStage::Suppressed}else{server_api::NotificationDiagnosticStage::Deferred});
}
assert_eq!(
job_notice_evidence(&store, &source),
(
@@ -534,6 +549,19 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and
let store = SqliteWorkspaceStore::open_for_test(path).unwrap();
let pending = store.list_pending_job_notifications("w", 100).unwrap();
assert_eq!(pending.len(), 4);
for (id,stage) in [(&accepted,server_api::NotificationDiagnosticStage::RuntimeAccepted),(&clean,server_api::NotificationDiagnosticStage::Suppressed),(&stale,server_api::NotificationDiagnosticStage::Suppressed),(&unknown,server_api::NotificationDiagnosticStage::AcceptanceUnknown),(&ambiguous,server_api::NotificationDiagnosticStage::Deferred)] {
let page=store.notification_diagnostics("w",&server_api::NotificationDiagnosticsQuery{delivery_id:Some(id.clone()),..Default::default()}).unwrap();
let item=&page.items[0];
assert_eq!(item.stage,stage);
assert!(item.legacy_evidence);
assert!(!item.model_processing_confirmed);
if id==&accepted {
assert!(!item.retry_candidate && !item.acceptance_outcome_unknown);
assert!(item.target_worker_id.is_some());
}
let json=serde_json::to_string(item).unwrap();
assert!(!json.contains("record_json") && !json.contains("notification_suppressed"));
}
let checker_body = store
.job_notification_request("w", &checker)
.unwrap()
crates/workspace-server/src/notification_diagnostics.rs
+7 -4
View File
@@ -157,14 +157,16 @@ pub(super) fn page_with_connection(
d.disposition,d.reason,d.purpose,d.delivery_id,
ar.delivery_id IS NOT NULL,
(r.delivery_id IS NOT NULL OR ap.delivery_id IS NOT NULL),
(a.delivery_id IS NOT NULL OR aa.delivery_id IS NOT NULL),
COALESCE(json_extract(a.target_json,'$.worker.runtime_id'),json_extract(aa.source_json,'$.runtime_id'),json_extract(r.target_json,'$.worker.runtime_id'),json_extract(ar.reserved_source_json,'$.runtime_id')),
COALESCE(json_extract(a.target_json,'$.worker.worker_id'),json_extract(aa.source_json,'$.worker_id'),json_extract(r.target_json,'$.worker.worker_id'),json_extract(ar.reserved_source_json,'$.worker_id')),
(a.delivery_id IS NOT NULL OR aa.delivery_id IS NOT NULL OR COALESCE(l.runtime_accepted,0)=1),
COALESCE(json_extract(a.target_json,'$.worker.runtime_id'),json_extract(aa.source_json,'$.runtime_id'),json_extract(r.target_json,'$.worker.runtime_id'),json_extract(r.target_json,'$.runtime_id'),json_extract(ar.reserved_source_json,'$.runtime_id'),json_extract(l.record_json,'$.target.runtime_id')),
COALESCE(json_extract(a.target_json,'$.worker.worker_id'),json_extract(aa.source_json,'$.worker_id'),json_extract(r.target_json,'$.worker.worker_id'),json_extract(r.target_json,'$.worker_id'),json_extract(ar.reserved_source_json,'$.worker_id'),json_extract(l.record_json,'$.target.worker_id')),
f.code,f.recorded_at,
CASE WHEN s.role IN ('orchestrator','worker') AND s.assignment_id!='' THEN EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=s.workspace_id AND c.ticket_id=s.ticket_id AND c.role=s.role AND c.assignment_id=s.assignment_id) END,
EXISTS(SELECT 1 FROM notification_failures uf WHERE uf.workspace_id=s.workspace_id AND uf.delivery_id=d.delivery_id AND uf.code IN ('acceptance_unknown','acknowledgement_failed')),
(SELECT COUNT(*) FROM (SELECT DISTINCT at.runtime_id,at.worker_id FROM notification_assignment_targets at WHERE at.workspace_id=s.workspace_id AND at.delivery_id=d.delivery_id))
(SELECT COUNT(*) FROM (SELECT DISTINCT at.runtime_id,at.worker_id FROM notification_assignment_targets at WHERE at.workspace_id=s.workspace_id AND at.delivery_id=d.delivery_id)),
l.delivery_id IS NOT NULL
{joins}
LEFT JOIN notification_job_legacy_receipts l ON l.workspace_id=s.workspace_id AND l.delivery_id=d.delivery_id
LEFT JOIN notification_requests r ON r.workspace_id=s.workspace_id AND r.delivery_id=d.delivery_id
LEFT JOIN notification_acknowledgements a ON a.workspace_id=s.workspace_id AND a.delivery_id=d.delivery_id
LEFT JOIN ticket_assignment_deliveries ar ON ar.workspace_id=s.workspace_id AND ar.delivery_id=d.delivery_id
@@ -249,6 +251,7 @@ pub(super) fn page_with_connection(
purpose: bounded(row.get(10)?),
delivery_id: bounded(row.get(11)?),
stage,
legacy_evidence: row.get(22)?,
acceptance_outcome_unknown,
target_resolution,
target_runtime_id,
crates/workspace-server/src/server_job_notification_tests.rs
+9
View File
@@ -51,6 +51,10 @@ async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_comm
let evidence = job_notification_evidence(&api, &id);
assert!(!evidence.2);
assert_eq!(evidence.3.as_deref(), Some("acceptance_unknown"));
let diagnostics=delivery_http(&api,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?job_id={}",request.job_id),None,StatusCode::OK).await;
assert_eq!(diagnostics["items"][0]["stage"],"acceptance_unknown");
assert_eq!(diagnostics["items"][0]["target_worker_id"],target.worker_id);
assert_eq!(diagnostics["items"][0]["acceptance_outcome_unknown"],true);
assert_eq!(delivery_receipts(&api, &target).len(), 1);
// Reopen only Backend persistence, retaining the real Runtime receipt/session.
let mut reopened = api.clone();
@@ -77,6 +81,11 @@ async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_comm
});
delivery_wait_idle(&reopened, &target, 1).await;
assert!(job_notification_evidence(&reopened, &id).2);
let diagnostics=delivery_http(&reopened,"GET",&format!("/api/w/{TEST_WORKSPACE_ID}/notification-diagnostics?job_id={}",request.job_id),None,StatusCode::OK).await;
assert_eq!(diagnostics["items"][0]["stage"],"runtime_accepted");
assert_eq!(diagnostics["items"][0]["target_worker_id"],target.worker_id);
assert_eq!(diagnostics["items"][0]["acceptance_outcome_unknown"],false);
assert_eq!(diagnostics["items"][0]["model_processing_confirmed"],false);
assert_eq!(delivery_receipts(&reopened, &target).len(), 1);
assert_eq!(delivery_commits(&reopened, &target), vec![pinned.0.clone()]);
let attempts = execution.attempts.lock().unwrap();