Skip to content

refactor(native-integration): make stores async, drop actors #4143

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ScriptedAlchemy merged 2 commits into from
Oct 16, 2026
Merged

refactor(native-integration): make stores async, drop actors #4143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a67ce51
refactor(native-integration): make store async, drop actor
ScriptedAlchemy Oct 16, 2026
87bcf02
refactor(stack-delivery): await delivery store, drop actor
ScriptedAlchemy Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 71 additions & 54 deletions crates/tracedecay-agent-hosts/src/native_integration/journey_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use super::analysis::DaemonNativeIntegrationAnalysisV1;
use super::registry::{DaemonNativeIntegrationServiceRegistry, NativeIntegrationTargetV1};
use super::stack_signals::signal_from_preflight;
use super::store::{DaemonNativeIntegrationStore, SharedDaemonNativeIntegrationStore};
use super::store::DaemonNativeIntegrationStore;
use super::worktree::DaemonNativeWorktreeAuthority;
use tracedecay_application::native_integration::{
GixNativeIntegrationAdapter, NativeApplyEffectV1, NativeIntegrationAnalysisPort,
Expand Down Expand Up @@ -533,11 +533,12 @@ async fn preflight(
request: NativeIntegrationPreflightRequestV1,
) -> tracedecay_domain::NativeIntegrationPreviewV1 {
let signal = CancellationSignal::active("cancel.native.journey.preflight").expect("signal");
let outcome =
native_runtime::spawn_blocking(move || owner.service().preflight(request, &signal))
.await
.expect("preflight join")
.expect("preflight result");
let outcome = native_runtime::on_blocking_pool(async move {
owner.service().preflight(request, &signal).await
})
.await
.expect("preflight join")
.expect("preflight result");
let NativeIntegrationPreflightOutcomeV1::Preview(preview) = outcome else {
panic!("fixture must produce an eligible native preview: {outcome:?}");
};
Expand Down Expand Up @@ -818,10 +819,12 @@ async fn checked_out_declared_stack_conflict_enqueues_without_moving_refs() {
owner
.store()
.save_preview(preview.clone())
.await
.expect("save canonical conflict preview");
owner
.store()
.save_preview(occupied_preview.clone())
.await
.expect("save canonical occupied preview");
let stack_runtime = mount_stack_runtime_off_tokio(
owner.clone(),
Expand Down Expand Up @@ -855,17 +858,21 @@ async fn checked_out_declared_stack_conflict_enqueues_without_moving_refs() {
.expect("other recipient signal");
stack_runtime
.enqueue_from_preflight(other_signal.clone(), &other_context)
.await
.expect("enqueue older other-recipient signal");
stack_runtime
.enqueue_from_preflight(signal.clone(), &live_context)
.await
.expect("enqueue actual conflict");
stack_runtime
.enqueue_from_preflight(occupied_signal.clone(), &live_context)
.await
.expect("enqueue destination-occupied dependency-ready signal");
tokio::time::timeout(Duration::from_secs(3), async {
loop {
if stack_runtime
.pending_host_deliveries()
.await
.is_ok_and(|pending| pending.len() == 3)
{
break;
Expand All @@ -878,14 +885,16 @@ async fn checked_out_declared_stack_conflict_enqueues_without_moving_refs() {

let cancellation = CancellationSignal::active("cancel.native.journey.declared-conflict.expand")
.expect("expand cancellation");
let denied = stack_runtime.expand(
GitHubStackSignalExpandSurfaceRequest {
signal_id: Some(signal.signal_id.clone()),
expected_watermark_id: Some(signal.watermark_id.clone()),
}
.into_application_request(other_context),
&cancellation,
);
let denied = stack_runtime
.expand(
GitHubStackSignalExpandSurfaceRequest {
signal_id: Some(signal.signal_id.clone()),
expected_watermark_id: Some(signal.watermark_id.clone()),
}
.into_application_request(other_context),
&cancellation,
)
.await;
assert_eq!(denied, Err(GitHubStackSignalExpandPortError::Concealed));
let expanded = stack_runtime
.expand(
Expand All @@ -896,6 +905,7 @@ async fn checked_out_declared_stack_conflict_enqueues_without_moving_refs() {
.into_application_request(live_context.clone()),
&cancellation,
)
.await
.expect("expand exact authorized conflict signal");
let GitHubStackSignalExpandSurfaceResultV1::Expanded { evidence } = expanded else {
panic!("expected expanded stack conflict evidence");
Expand Down Expand Up @@ -931,6 +941,7 @@ async fn checked_out_declared_stack_conflict_enqueues_without_moving_refs() {
.into_application_request(live_context),
&cancellation,
)
.await
.expect("expand destination-occupied dependency-ready signal");
let GitHubStackSignalExpandSurfaceResultV1::Expanded { evidence } = occupied else {
panic!("expected destination-occupied stack evidence");
Expand Down Expand Up @@ -1158,6 +1169,7 @@ async fn independent_pair_applies_supported_modes_and_survives_daemon_restart()
owner
.store()
.save_approval(approval.clone())
.await
.expect("durably issue approval");
let transaction_id =
NativeIntegrationTransactionId::new(format!("transaction.native.journey.{index}"))
Expand All @@ -1166,17 +1178,20 @@ async fn independent_pair_applies_supported_modes_and_survives_daemon_restart()
let apply_owner = owner.clone();
let apply_signal =
CancellationSignal::active("cancel.native.journey.apply").expect("signal");
let receipt = native_runtime::spawn_blocking(move || {
apply_owner.service().apply(
NativeIntegrationApplyRequestV1 {
context,
transaction_id: transaction_id.clone(),
preview,
approval,
observed_at: OBSERVED_AT,
},
&apply_signal,
)
let receipt = native_runtime::on_blocking_pool(async move {
apply_owner
.service()
.apply(
NativeIntegrationApplyRequestV1 {
context,
transaction_id: transaction_id.clone(),
preview,
approval,
observed_at: OBSERVED_AT,
},
&apply_signal,
)
.await
})
.await
.expect("apply join")
Expand Down Expand Up @@ -1294,11 +1309,13 @@ async fn independent_pair_applies_supported_modes_and_survives_daemon_restart()
.await;
stack_runtime
.enqueue_from_preflight(stack_signal.clone(), &live_context)
.await
.expect("enqueue terminal signal");
tokio::time::timeout(Duration::from_secs(3), async {
loop {
if stack_runtime
.pending_host_deliveries()
.await
.is_ok_and(|pending| pending.len() == 1)
{
break;
Expand All @@ -1318,6 +1335,7 @@ async fn independent_pair_applies_supported_modes_and_survives_daemon_restart()
&CancellationSignal::active("cancel.native.journey.terminal-consumer")
.expect("cancellation"),
)
.await
.expect("expand terminal signal");
let GitHubStackSignalExpandSurfaceResultV1::Expanded { evidence } = expanded else {
panic!("expected terminal stack evidence");
Expand Down Expand Up @@ -1355,17 +1373,14 @@ async fn independent_pair_applies_supported_modes_and_survives_daemon_restart()
registry.shutdown().await.expect("shutdown owner registry");
let (restarted_registry, restarted_owner, _analysis) =
mount(database, repository_root).await;
let durable = native_runtime::spawn_blocking(move || {
restarted_owner.service().status(
tracedecay_contracts::NativeIntegrationStatusRequestV1 {
transaction_id: status_transaction_id,
},
)
})
.await
.expect("status join")
.expect("durable status")
.expect("durable transaction status");
let durable = restarted_owner
.service()
.status(tracedecay_contracts::NativeIntegrationStatusRequestV1 {
transaction_id: status_transaction_id,
})
.await
.expect("durable status")
.expect("durable transaction status");
assert_eq!(
durable.terminal_outcome,
Some(NativeIntegrationTerminalOutcomeV1::Committed)
Expand Down Expand Up @@ -1450,6 +1465,7 @@ async fn foreign_destination_ref_drift_terminates_without_mutating_the_foreign_t
owner
.store()
.save_approval(approval.clone())
.await
.expect("durably issue approval");

git(&repository_root, &["checkout", "-b", "foreign"]);
Expand All @@ -1468,20 +1484,23 @@ async fn foreign_destination_ref_drift_terminates_without_mutating_the_foreign_t

let apply_owner = owner.clone();
let signal = CancellationSignal::active("cancel.native.journey.drift").expect("signal");
let receipt = native_runtime::spawn_blocking(move || {
apply_owner.service().apply(
NativeIntegrationApplyRequestV1 {
context,
transaction_id: NativeIntegrationTransactionId::new(
"transaction.native.journey.drift",
)
.expect("transaction id"),
preview,
approval,
observed_at: OBSERVED_AT,
},
&signal,
)
let receipt = native_runtime::on_blocking_pool(async move {
apply_owner
.service()
.apply(
NativeIntegrationApplyRequestV1 {
context,
transaction_id: NativeIntegrationTransactionId::new(
"transaction.native.journey.drift",
)
.expect("transaction id"),
preview,
approval,
observed_at: OBSERVED_AT,
},
&signal,
)
.await
})
.await
.expect("apply join")
Expand Down Expand Up @@ -1566,9 +1585,7 @@ async fn owned_worktree_inspection_shares_the_discovery_walk_another_thread_owns
)
.expect("registered authorized scope set"),
);
let store = SharedDaemonNativeIntegrationStore::from_arc(Arc::new(
DaemonNativeIntegrationStore::open(database).expect("native integration store"),
));
let store = DaemonNativeIntegrationStore::open(database);
let authority = Arc::new(
DaemonNativeWorktreeAuthority::open(
project_id.clone(),
Expand Down Expand Up @@ -1620,5 +1637,5 @@ async fn owned_worktree_inspection_shares_the_discovery_walk_another_thread_owns
1
);
tracedecay_runtime_core::git_repository::reset_repository_discovery_for_test(&repository_root);
store.shutdown().expect("shutdown native integration store");
assert!(store.shutdown(), "the store closes once");
}
5 changes: 2 additions & 3 deletions crates/tracedecay-agent-hosts/src/native_integration/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
//! Daemon composition of the Plan 36 native-integration authority.
//!
//! `registry` retains one composed transaction coordinator per exact
//! project/repository identity; `store` bridges the synchronous
//! `tracedecay-store` contract onto the async registered session database
//! through one bounded actor per database.
//! project/repository identity; `store` implements the async
//! `tracedecay-store` contract directly over the registered session database.

mod analysis;
mod registry;
Expand Down
Loading
Loading