chore: merge develop into hare/develop

This commit is contained in:
2026-10-17 12:06:05 +09:00
100 changed files with 12102 additions and 958 deletions
+2 -1
View File
@@ -53,7 +53,8 @@ ts-rs = "12.0.1"
url.workspace = true
uuid = { workspace = true, features = ["v7"] }
zeroize.workspace = true
webauthn-rs = { workspace = true }
webauthn-rs = { workspace = true, features = ["conditional-ui"] }
webauthn-rs-proto = "0.5.5"
[dev-dependencies]
wip-protocol.workspace = true
@@ -529,7 +529,7 @@ fn activation_oldest_and_previous_schema_migrate_with_same_constraints_and_prese
assert!(!store.is_user_pending_activation("old-user").unwrap());
store.with_conn(|conn| {
assert_eq!(schema(conn), expected);
assert_eq!(current_schema_version(conn)?, 100);
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
assert!(conn.execute("INSERT INTO user_activations VALUES('missing','issue','hash','future',0)", []).is_err());
for attempts in [-1, 6] { assert!(conn.execute("INSERT INTO user_activations VALUES('old-user','issue','hash','future',?1)", [attempts]).is_err()); }
Ok(())
@@ -474,7 +474,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
(97, "MR notifications and authenticated implementation results".into()),
(98, "Assignment-independent implementation reports".into()),
(99, "Explicit hosting administrators and self-registration policy".into()),
(100, "Provisional users and scoped passkey activation".into())
(100, "Provisional users and scoped passkey activation".into()),
(101, "Durable explicit Subject body release".into())
]
);
for (table, rows) in retained.iter().zip(original.iter()) {
@@ -282,6 +282,10 @@ pub struct SubjektivConsolidationGrant {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct SubjektivJournalGrant {
/// Frozen writer generation. Legacy absence preserves its original canonical wire/digest;
/// recovery must use only the immutable original Session and still-current body.
#[serde(default, skip_serializing_if = "String::is_empty")]
pub binding_id: String,
pub subject_id: String,
pub session_id: String,
pub journal_id: String,
+51 -1
View File
@@ -1288,6 +1288,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
})
}
fn subject_release(
&self,
_request: runtime_api::SubjectReleaseRequest,
) -> Result<runtime_api::SubjectReleaseResponse, String> {
Err("actual Subject Controller reservation is unavailable on this Runtime".into())
}
fn session_material_pin(
&self,
_request: runtime_api::SessionMaterialPinRequest,
@@ -2244,6 +2251,21 @@ impl RuntimeRegistry {
runtime.worker_session_attachment(workspace_id, &worker_ref, session_id, attachment_id)
}
pub fn subject_release(
&self,
runtime_id: &str,
request: runtime_api::SubjectReleaseRequest,
) -> Result<runtime_api::SubjectReleaseResponse, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
self.runtime(runtime_id)?
.subject_release(request)
.map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: runtime_id.into(),
code: "subject_release_unproven".into(),
message,
})
}
pub fn session_material_pin(
&self,
runtime_id: &str,
@@ -3297,6 +3319,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
})
}
fn subject_release(
&self,
request: runtime_api::SubjectReleaseRequest,
) -> Result<runtime_api::SubjectReleaseResponse, String> {
if request.workspace_id != self.workspace_id {
return Err("Subject release Workspace mismatch".into());
}
let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend");
self.runtime
.subject_release_scoped(&scope, request)
.map_err(|e| e.to_string())
}
fn session_material_pin(
&self,
request: runtime_api::SessionMaterialPinRequest,
@@ -4114,7 +4149,10 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s
{
return "workers:input";
}
if path.ends_with("/stop") || path.ends_with("/cancel") {
if path.ends_with("/stop")
|| path.ends_with("/cancel")
|| (path == "/v1/subject-release" && method == "POST")
{
return "workers:stop";
}
if path == "/v1/protocol/ws" {
@@ -5523,6 +5561,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
})
}
fn subject_release(
&self,
request: runtime_api::SubjectReleaseRequest,
) -> Result<runtime_api::SubjectReleaseResponse, String> {
self.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move { client.subject_release(request).await },
)
.map_err(|diagnostic| diagnostic.message)
}
fn session_material_pin(
&self,
request: runtime_api::SessionMaterialPinRequest,
@@ -381,6 +381,8 @@ async fn implementation_result_v97_to_v98_preserves_legacy_bytes_receipts_and_no
DROP TABLE ticket_implementation_results;
DROP TABLE ticket_implementation_delegations;
DROP TABLE merge_request_source_observations;
DROP TABLE subject_worker_releases;
UPDATE __yoi_schema_migrations SET version=100 WHERE version=101;
DROP TABLE user_activations;
UPDATE __yoi_schema_migrations SET version=99 WHERE version=100;
DROP TABLE hosting_administrators; DROP TABLE hosting_settings;
@@ -513,6 +515,8 @@ async fn mr_notification_upgrade_recovers_events_not_comments_and_retains_unknow
// 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 IF EXISTS implementation_delegation_capture;
DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations;
DROP TABLE subject_worker_releases;
UPDATE __yoi_schema_migrations SET version=100 WHERE version=101;
DROP TABLE user_activations;
UPDATE __yoi_schema_migrations SET version=99 WHERE version=100;
DROP TABLE hosting_administrators; DROP TABLE hosting_settings;
@@ -671,7 +675,9 @@ async fn mr_event_actors_preserve_self_suppression_in_capture_and_backfill() {
if historical {
conn.execute_batch("DROP TRIGGER notification_mr_event; DROP TRIGGER notification_mr_source; DROP TRIGGER notification_implementation_result; DROP TRIGGER IF EXISTS implementation_delegation_capture;
DROP TABLE ticket_implementation_results; DROP TABLE ticket_implementation_delegations; DROP TABLE merge_request_source_observations;
DROP TABLE user_activations;
DROP TABLE subject_worker_releases;
UPDATE __yoi_schema_migrations SET version=100 WHERE version=101;
DROP TABLE user_activations;
UPDATE __yoi_schema_migrations SET version=99 WHERE version=100;
DROP TABLE hosting_administrators; DROP TABLE hosting_settings;
UPDATE __yoi_schema_migrations SET version=98 WHERE version=99;
@@ -699,6 +699,22 @@ CREATE TABLE worker_create_reservations (
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
CHECK ((singleton_key IS NULL) = (singleton_generation IS NULL))
);
CREATE TABLE subject_worker_releases (
workspace_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
singleton_key TEXT NOT NULL,
singleton_generation INTEGER NOT NULL CHECK (singleton_generation > 0),
binding_id TEXT NOT NULL,
operation_id TEXT NOT NULL,
proof TEXT CHECK (proof IN ('runtime_disconnected','erased')),
completed INTEGER NOT NULL DEFAULT 0 CHECK (completed IN (0,1)),
cancelled INTEGER NOT NULL DEFAULT 0 CHECK (cancelled IN (0,1)),
PRIMARY KEY (workspace_id, operation_id),
CHECK (cancelled = 0 OR (completed = 0 AND proof IS NULL)),
FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_create_reservations(workspace_id, worker_id) ON DELETE CASCADE
);
CREATE UNIQUE INDEX subject_worker_release_active ON subject_worker_releases(workspace_id,worker_id) WHERE cancelled=0;
CREATE TRIGGER worker_create_reservation_singleton_lease_insert
BEFORE INSERT ON worker_create_reservations
WHEN (NEW.singleton_key IS NULL AND NEW.singleton_generation IS NOT NULL)
@@ -57,6 +57,7 @@ fn fresh_and_upgraded_state_contract_schemas_enforce_the_same_constraints() {
"workspace_config_tree_history",
"workspace_memory_settings",
"worker_create_reservations",
"subject_worker_releases",
"backend_jobs",
"backend_job_attempts",
"flow_sources",
+361 -197
View File
@@ -28,6 +28,7 @@ use subjektiv::api::*;
mod backend_job_profile;
mod subjektiv_job;
mod subjektiv_journal;
mod subjektiv_memory_journal;
mod worker_operations;
mod worker_protocol;
pub(crate) use worker_operations::{
@@ -1654,6 +1655,7 @@ pub struct WorkspaceApi {
Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<Mutex<OrchestratorAttentionState>>>>>,
notification_delivery_lifecycle: Arc<notification_delivery::NotificationDeliveryLifecycle>,
backend_job_drain_requests: Arc<AtomicU64>,
journal_lifecycle: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
journal_dispatch_owner: Option<Arc<WorkspaceApi>>,
observation_proxy: BackendObservationProxy,
runtime_subscription_broker: RuntimeSubscriptionBroker,
@@ -3379,6 +3381,7 @@ fn workspace_request_is_callback(method: &Method, path: &str, workspace_id: &str
|| path == format!("/api/runtime/v1/workspaces/{workspace_id}/resources/fetch")))
|| (*method == Method::GET
&& (path == format!("/api/w/{workspace_id}/runtime-config")
|| path == format!("/api/w/{workspace_id}/subjektiv/journal/startup")
|| path == format!("/api/w/{workspace_id}/worker-discovery/workers")))
}
@@ -4160,6 +4163,7 @@ impl WorkspaceApi {
notification_delivery_lifecycle: Arc::new(Default::default()),
orchestrator_attention: Arc::new(Mutex::new(HashMap::new())),
backend_job_drain_requests: Arc::new(AtomicU64::new(0)),
journal_lifecycle: Arc::new(Mutex::new(HashMap::new())),
journal_dispatch_owner: None,
observation_proxy,
runtime_subscription_broker,
@@ -4362,8 +4366,9 @@ impl WorkspaceApi {
resolved_worker_observation_grants: Vec::new(),
resolved_workspace_api: None,
resolved_memory_settings: None,
resolved_subjektiv_attached: request.grants.subjektiv_consolidation.is_some()
|| request.grants.subjektiv_journal.is_some(),
// Journal Jobs use their independent frozen Job Host grant, not the
// main-Subject/Memory attachment or its resident/extraction policy.
resolved_subjektiv_attached: request.grants.subjektiv_consolidation.is_some(),
resolved_control_operation: Some(control_operation),
},
) {
@@ -5468,6 +5473,22 @@ impl WorkspaceApi {
.map(|assignment| assignment.operation_id.clone())
})
.unwrap_or_else(|| format!("manual:{}", WorkerId::now_v7()));
let subject = singleton_key
.as_deref()
.and_then(|key| key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX));
let subject_gate = subject
.map(|s| subjektiv_journal::lifecycle_lock(self, s))
.transpose()?;
let subject_guard = subject_gate
.as_ref()
.map(|gate| {
gate.lock()
.map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))
})
.transpose()?;
if let Some(subject) = subject {
subjektiv_journal::check_start_allowed(self, subject)?;
}
let reservation = match self.config_store.reserve_worker_create(
&self.config.workspace_id,
runtime_id,
@@ -5490,6 +5511,7 @@ impl WorkspaceApi {
return Err(ApiError::with_diagnostics(error, diagnostics));
}
};
drop(subject_guard);
let worker_id = reservation.worker_id;
if let Some(assignment) = request.ticket_assignment.as_ref() {
self.store.bind_ticket_assignment_operation_worker(
@@ -5803,6 +5825,9 @@ impl WorkspaceApi {
diagnostics,
));
}
if let Err(error) = subjektiv_journal::reconcile_created_worker(self, &worker_ref) {
tracing::warn!(%error, worker_id=%worker_ref.worker_id, "Journal first-run reconciliation remains recoverable; saved history retained");
}
Ok(result)
}
@@ -13122,7 +13147,10 @@ fn build_inner_router(api: WorkspaceApi) -> Router {
"/api/w/{workspace_id}/workers/self/backend-job-result",
post(scoped_submit_backend_job_result),
)
.route("/api/w/{workspace_id}/subjektiv/journal/release", post(subjektiv_journal::release_route))
.route("/api/w/{workspace_id}/subjektiv/journal/startup", get(subjektiv_journal::startup_route))
.route("/api/w/{workspace_id}/subjektiv/journal/recording", post(subjektiv_journal::recording_route))
.route("/api/w/{workspace_id}/subjektiv/journal/memory", post(subjektiv_memory_journal::route))
.route("/api/w/{workspace_id}/subjektiv/journal/job", post(subjektiv_journal::job_route))
.route("/api/w/{workspace_id}/subjektiv/journal/job/finish", post(subjektiv_journal::finish_route))
.fallback(get(static_or_spa_fallback))
@@ -22351,6 +22379,7 @@ async fn scoped_record_subjektiv_session(
&request.session_id,
)?
};
subjektiv_journal::bind_host_session(&api, attribution.clone())?;
Ok(Json(server_api::SubjektivRecordSessionResponse {
subject_id: attribution.subject_id,
session_id: attribution.session_id,
@@ -29670,11 +29699,13 @@ struct PasskeyRegistrationEnvelope {
}
/// An account gets one stable WebAuthn user handle, without storing a new schema
/// field. Migration leaves all existing credentials, their random registration
/// handles and key material unchanged. Credential-ID-scoped authentication must
/// accept signed assertions both without userHandle and with the original random
/// or newly registered stable userHandle; it is not discoverable-handle lookup.
/// The real-verifier compatibility tests exercise each representation explicitly.
/// field. Legacy identified-account login accepts original random handles or no
/// returned handle, always requiring the saved credential's signature. Normal
/// discoverable login additionally requires the saved owner's stable handle.
/// Random registration handles were never stored, so they cannot be recovered
/// or safely mapped by trusting the unsigned assertion. Preserve these keys and
/// use explicit legacy login to authenticate before adding a stable-handle RK.
/// Merely having a stable handle does not prove the old key is discoverable.
fn passkey_user_handle(account_id: &str) -> Uuid {
let digest = Sha256::digest(account_id.as_bytes());
Uuid::from_bytes(
@@ -29778,7 +29809,7 @@ fn start_passkey_registration(
let header_origin = request_origin(headers);
let auth = auth_config_for_origin(&api.config, browser_origin.or(header_origin.as_deref()))?;
let webauthn = webauthn_for_auth(&auth)?;
let (public_key, verifier) = webauthn
let (mut public_key, verifier) = webauthn
.start_passkey_registration(
passkey_user_handle(&user.account_id),
&user.handle,
@@ -29786,6 +29817,21 @@ fn start_passkey_registration(
Some(exclude_credentials),
)
.map_err(|error| auth_error("webauthn_registration_options_failed", &error.to_string()))?;
// webauthn-rs passkeys require UV but do not request residency by default.
// These public protocol fields request an RK on conforming clients; residency
// itself is not remotely attestable. Keep the library's UV/verifier state.
let selection = public_key
.public_key
.authenticator_selection
.as_mut()
.ok_or_else(|| {
auth_error(
"webauthn_registration_options_failed",
"authenticator selection is missing",
)
})?;
selection.resident_key = Some(webauthn_rs_proto::ResidentKeyRequirement::Required);
selection.require_resident_key = true;
let challenge_id = new_id("webauthn-registration");
// New users do not exist yet, so the nullable FK remains empty. The complete
// target (including its allocated user ID) is inside the purpose-tagged state.
@@ -29950,6 +29996,18 @@ fn complete_passkey_enrollment(
issue_browser_session(api, user)
}
/// Normal login has no preselected identity. Explicit handle requests are the
/// legacy migration path, not an automatic fallback. Non-resident keys and keys
/// with old random user handles can authenticate there, then add an RK from the
/// live account. No saved key or owner is rewritten. Existing stable-handle RKs
/// can already use normal login; discoverability cannot be inferred from storage.
#[derive(Serialize, Deserialize)]
#[serde(tag = "mode", content = "verifier", rename_all = "snake_case")]
enum PasskeyLoginState {
Discoverable(webauthn_rs::prelude::DiscoverableAuthentication),
Legacy(PasskeyAuthentication),
}
async fn post_passkey_login_options(
State(api): State<ServerAuthApi>,
headers: HeaderMap,
@@ -29958,48 +30016,66 @@ async fn post_passkey_login_options(
api.store
.db_executor()
.run(move || {
let user = match request.handle.as_deref() {
Some(handle) => api.store.get_user_by_handle(&normalize_handle(handle)?)?,
None => {
return Err(auth_error(
"passkey_handle_required",
"login requires an explicit handle",
)
.into());
}
}
.ok_or_else(|| auth_error("unknown_auth_user", "no matching user account exists"))?;
if api.store.is_user_pending_activation(&user.user_id)? {
return Err(auth_error(
"passkey_activation_required",
"initial passkey activation is required",
)
.into());
}
let passkeys = passkeys_for_user(&api, &user.user_id)?;
if passkeys.is_empty() {
return Err(auth_error(
"passkey_not_registered",
"user has no registered passkey credentials",
)
.into());
}
let header_origin = request_origin(&headers);
let requested_origin = request
.browser_origin
.as_deref()
.or(header_origin.as_deref());
let auth = auth_config_for_origin(&api.config, requested_origin)?;
let auth = auth_config_for_origin(
&api.config,
request
.browser_origin
.as_deref()
.or(header_origin.as_deref()),
)?;
let webauthn = webauthn_for_auth(&auth)?;
let (public_key, state) = webauthn
.start_passkey_authentication(&passkeys)
.map_err(|error| auth_error("webauthn_login_options_failed", &error.to_string()))?;
let (public_key, state, user_id) =
if let Some(handle) = request.handle.as_deref() {
let user = api
.store
.get_user_by_handle(&normalize_handle(handle)?)?
.ok_or_else(|| {
auth_error("unknown_auth_user", "no matching user account exists")
})?;
if api.store.is_user_pending_activation(&user.user_id)? {
return Err(auth_error(
"passkey_activation_required",
"initial passkey activation is required",
)
.into());
}
let passkeys = passkeys_for_user(&api, &user.user_id)?;
if passkeys.is_empty() {
return Err(auth_error(
"passkey_not_registered",
"user has no registered passkey credentials",
)
.into());
}
let (options, state) = webauthn
.start_passkey_authentication(&passkeys)
.map_err(|error| {
auth_error("webauthn_login_options_failed", &error.to_string())
})?;
(
options,
PasskeyLoginState::Legacy(state),
Some(user.user_id),
)
} else {
let (mut options, state) = webauthn
.start_discoverable_authentication()
.map_err(|error| {
auth_error("webauthn_login_options_failed", &error.to_string())
})?;
// The feature exposes a conditional-UI builder. Use its empty allow
// list and required UV verifier with an explicit user gesture, not
// conditional/autofill mediation. No account inventory is returned.
options.mediation = None;
(options, PasskeyLoginState::Discoverable(state), None)
};
let challenge_id = new_id("webauthn-login");
api.store.put_auth_challenge(&AuthChallengeRecord {
challenge_id: challenge_id.clone(),
ceremony: "passkey_login".to_string(),
ceremony: "passkey_login".into(),
challenge: challenge_id.clone(),
user_id: Some(user.user_id),
user_id,
rp_id: auth.rp_id,
origin: auth.origin,
state_json: Some(serde_json::to_string(&state).map_err(|error| {
@@ -30041,64 +30117,114 @@ fn complete_passkey_login(
)
.into());
}
let user_id = challenge.user_id.clone().ok_or_else(|| {
auth_error(
"invalid_passkey_challenge",
"passkey login challenge is not bound to a user",
)
})?;
let user = api
.store
.get_user(&user_id)?
.ok_or_else(|| auth_error("unknown_auth_user", "passkey user does not exist"))?;
let state_json = challenge.state_json.clone().ok_or_else(|| {
auth_error(
"missing_webauthn_state",
"passkey login state was not persisted",
)
})?;
let state: PasskeyAuthentication = serde_json::from_str(&state_json)
.map_err(|error| auth_error("webauthn_state_deserialize_failed", &error.to_string()))?;
let webauthn = webauthn_for_challenge(&api.config, &challenge)?;
let auth_result = webauthn
.finish_passkey_authentication(&request.credential, &state)
.map_err(|error| auth_error("webauthn_login_verification_failed", &error.to_string()))?;
let credential_id = serde_json::to_value(auth_result.cred_id())
.ok()
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.ok_or_else(|| {
let state: PasskeyLoginState =
serde_json::from_str(challenge.state_json.as_deref().ok_or_else(|| {
auth_error(
"invalid_webauthn_credential_id",
"verified credential id was not serializable",
"missing_webauthn_state",
"passkey login state was not persisted",
)
})?;
})?)
.map_err(|error| auth_error("webauthn_state_deserialize_failed", &error.to_string()))?;
let credential_id = serde_json::to_value(webauthn_rs::prelude::Base64UrlSafeData::from(
request.credential.get_credential_id().to_vec(),
))
.ok()
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.ok_or_else(|| auth_error("invalid_webauthn_credential_id", "credential id is invalid"))?;
let stored = api
.store
.get_passkey_credential(&credential_id)?
.ok_or_else(|| {
auth_error(
"unknown_passkey_credential",
"verified passkey credential is not registered",
"passkey credential is not registered",
)
})?;
if stored.user_id != user.user_id {
let user = api
.store
.get_user(&stored.user_id)?
.ok_or_else(|| auth_error("unknown_auth_user", "passkey user does not exist"))?;
if api.store.is_user_pending_activation(&user.user_id)? {
return Err(auth_error(
"passkey_user_mismatch",
"verified passkey credential does not belong to the challenged user",
"passkey_activation_required",
"initial passkey activation is required",
)
.into());
}
let mut passkey: Passkey = serde_json::from_str(&stored.public_key_cose)
.map_err(|error| auth_error("webauthn_passkey_deserialize_failed", &error.to_string()))?;
if passkey_credential_id(&passkey)? != credential_id {
return Err(auth_error(
"passkey_user_mismatch",
"stored credential id does not match key material",
)
.into());
}
let webauthn = webauthn_for_challenge(&api.config, &challenge)?;
let auth_result = match state {
PasskeyLoginState::Discoverable(state) => {
// userHandle is unsigned routing data, never identity authority. The
// candidate owner comes only from the exact saved credential ID;
// its stable handle must agree, and the saved key must verify below.
if challenge.user_id.is_some()
|| request.credential.get_user_unique_id()
!= Some(passkey_user_handle(&user.account_id).as_bytes().as_slice())
{
return Err(auth_error(
"passkey_user_mismatch",
"credential owner/user handle mismatch; use explicit legacy login for old keys",
)
.into());
}
webauthn.finish_discoverable_authentication(
&request.credential,
state,
&[passkey.clone().into()],
)
}
PasskeyLoginState::Legacy(state) => {
if challenge.user_id.as_deref() != Some(user.user_id.as_str()) {
return Err(auth_error(
"passkey_user_mismatch",
"credential does not belong to the challenged user",
)
.into());
}
webauthn.finish_passkey_authentication(&request.credential, &state)
}
}
.map_err(|error| auth_error("webauthn_login_verification_failed", &error.to_string()))?;
if auth_result.cred_id().as_slice() != request.credential.get_credential_id() {
return Err(auth_error(
"passkey_user_mismatch",
"verified credential differs from saved credential",
)
.into());
}
if stored.sign_count > 0
&& auth_result.counter() > 0
&& u64::from(auth_result.counter()) <= stored.sign_count
{
return Err(auth_error(
"webauthn_login_verification_failed",
"credential counter did not advance",
)
.into());
}
passkey.update_credential(&auth_result);
let public_key_cose = serde_json::to_string(&passkey)
.map_err(|error| auth_error("webauthn_state_serialize_failed", &error.to_string()))?;
api.store
.upsert_passkey_credential(&PasskeyCredentialRecord {
credential_id,
user_id: user.user_id.clone(),
public_key_cose: stored.public_key_cose,
public_key_cose,
transports_json: stored.transports_json,
sign_count: u64::from(auth_result.counter()),
created_at: stored.created_at,
last_used_at: Some(crate::auth::now_rfc3339()),
})?;
issue_browser_session(&api, user)
issue_browser_session(api, user)
}
fn issue_browser_session(
@@ -36815,6 +36941,7 @@ mod tests {
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
mod subjektiv_journal_tests;
mod subjektiv_memory_journal_tests;
mod ticket_evidence_tests;
mod ticket_worker_claim_tests;
mod ticket_worker_launch_tests;
@@ -46462,8 +46589,8 @@ mod tests {
let runtime = WorkdirlessFixtureRuntime::default();
api.runtime.register_or_replace(runtime.clone());
let request = |operation_id: &str| WorkerSpawnRequest {
requested_worker_name: Some("Subject Worker".to_string()),
singleton_key: Some("subjektiv:subject-retry".to_string()),
requested_worker_name: Some("Generic singleton Worker".to_string()),
singleton_key: Some("fixture:subject-retry".to_string()),
intent: WorkerSpawnIntent::WorkspaceCompanion,
acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
expected_segments: 0,
@@ -46500,7 +46627,7 @@ mod tests {
// Runtime dispatch. The restarted request receives a fresh manual allocation key, yet its
// identical singleton intent replays the reserved Worker and dispatches that generation.
let mut restart_request = request("ignored-manual-operation");
restart_request.singleton_key = Some("subjektiv:restart-retry".to_string());
restart_request.singleton_key = Some("fixture:restart-retry".to_string());
restart_request.resolved_control_operation = None;
let restart_fingerprint = worker_spawn_create_fingerprint(&restart_request).unwrap();
let memory = api
@@ -46547,7 +46674,7 @@ mod tests {
assert_eq!(rejected.state, InternalWorkerOperationState::Rejected);
assert!(
api.store
.current_worker_singleton_owner(&api.config.workspace_id, "subjektiv:subject-retry")
.current_worker_singleton_owner(&api.config.workspace_id, "fixture:subject-retry")
.unwrap()
.is_none()
);
@@ -46561,14 +46688,14 @@ mod tests {
let worker = created.worker.unwrap();
let current_owner = api
.store
.current_worker_singleton_owner(&api.config.workspace_id, "subjektiv:subject-retry")
.current_worker_singleton_owner(&api.config.workspace_id, "fixture:subject-retry")
.unwrap()
.unwrap();
assert_eq!(current_owner.worker, worker.worker);
assert_eq!(current_owner.generation, 3);
assert_eq!(
worker.singleton_key,
Some("subjektiv:subject-retry".to_string())
Some("fixture:subject-retry".to_string())
);
// Idle and paused Runtime metadata never releases Backend ownership, and a conflict is
@@ -46598,7 +46725,7 @@ mod tests {
assert_eq!(restored.state, server_api::WorkerRestoreState::Accepted);
assert_eq!(
restored.worker.unwrap().singleton_key.as_deref(),
Some("subjektiv:subject-retry")
Some("fixture:subject-retry")
);
// Replacing the connected Runtime with an empty instance models an unavailable/restarted
@@ -46618,10 +46745,7 @@ mod tests {
);
assert_eq!(
api.store
.current_worker_singleton_owner(
&api.config.workspace_id,
"subjektiv:subject-retry",
)
.current_worker_singleton_owner(&api.config.workspace_id, "fixture:subject-retry",)
.unwrap()
.unwrap()
.worker,
@@ -46698,10 +46822,7 @@ mod tests {
));
assert_eq!(
api.store
.current_worker_singleton_owner(
&api.config.workspace_id,
"subjektiv:subject-retry",
)
.current_worker_singleton_owner(&api.config.workspace_id, "fixture:subject-retry",)
.unwrap()
.unwrap()
.worker,
@@ -51676,16 +51797,20 @@ mod tests {
}
/// Generated HTTP admission, real SQLite and WebAuthn verification are connected
/// here. Only the physical authenticator is replaced with SoftPasskey; all
/// here. The authenticator is a resident software model with a SoftPasskey signing engine; all
/// identities used for addition come from cookies/tokens issued by real login.
mod passkey_security {
use super::*;
use webauthn_authenticator_rs::{WebauthnAuthenticator, softpasskey::SoftPasskey};
mod resident_model {
include!("../../../tests/e2e/src/resident_authenticator.rs");
}
use resident_model::ResidentAuthenticator;
include!("server_activation_tests.rs");
include!("server_discoverable_tests.rs");
const ORIGIN_URL: &str = "https://workspace.test";
type Authenticator = WebauthnAuthenticator<SoftPasskey>;
type Authenticator = ResidentAuthenticator;
struct Fixture {
_root: tempfile::TempDir,
@@ -52013,7 +52138,7 @@ mod tests {
}
fn authenticator() -> Authenticator {
WebauthnAuthenticator::new(SoftPasskey::new(true))
ResidentAuthenticator::new(true)
}
fn rejected(result: (StatusCode, HeaderMap, Value), status: StatusCode, code: &str) {
@@ -52204,17 +52329,17 @@ mod tests {
.unwrap()
.is_none()
);
rejected(
f.post(
let (status, _, options) = f
.post(
"/api/auth/passkeys/login/options",
None,
Some(ORIGIN_URL),
serde_json::json!({}),
)
.await,
StatusCode::BAD_REQUEST,
"passkey_handle_required",
);
.await;
assert_eq!(status, StatusCode::OK, "{options}");
let options: PasskeyLoginOptionsResponse = serde_json::from_value(options).unwrap();
assert!(options.public_key.public_key.allow_credentials.is_empty());
assert_eq!(f.credentials(&user), originals);
}
@@ -53059,113 +53184,152 @@ mod tests {
#[tokio::test]
async fn legacy_random_user_handle_credential_authenticates_before_and_after_stable_handle_addition()
{
let f = Fixture::new().await;
let api = ServerAuthApi::from(&f.api);
let user = allocate_new_user(&api, "legacy", None).unwrap();
let random_handle = Uuid::now_v7();
assert_ne!(random_handle, passkey_user_handle(&user.account_id));
let webauthn = webauthn_for_auth(&auth_public_config(&api.config)).unwrap();
let (options, state) = webauthn
.start_passkey_registration(random_handle, &user.handle, &user.display_name, None)
.unwrap();
let legacy_user_handle = options.public_key.user.id.clone();
assert_eq!(legacy_user_handle.as_slice(), random_handle.as_bytes());
let mut legacy_authenticator = authenticator();
let credential = legacy_authenticator
.do_registration(Url::parse(ORIGIN_URL).unwrap(), options)
.unwrap();
let passkey = webauthn
.finish_passkey_registration(&credential, &state)
.unwrap();
let record = PasskeyCredentialRecord {
credential_id: passkey_credential_id(&passkey).unwrap(),
user_id: user.user_id.clone(),
public_key_cose: serde_json::to_string(&passkey).unwrap(),
transports_json: None,
sign_count: 0,
created_at: crate::auth::now_rfc3339(),
last_used_at: None,
};
f.api
.store
.register_verified_passkey(
&PasskeyRegistrationTarget::NewAccount { user: user.clone() },
&record,
)
.unwrap();
let expected = user_response(user.clone());
for returned_handle in [None, Some(legacy_user_handle.as_slice())] {
let (_, logged_in) = f
.login_session_with_user_handle(
"legacy",
&mut legacy_authenticator,
returned_handle,
for resident in [false, true] {
let f = Fixture::new().await;
let api = ServerAuthApi::from(&f.api);
let user = allocate_new_user(&api, "legacy", None).unwrap();
let random_handle = Uuid::now_v7();
assert_ne!(random_handle, passkey_user_handle(&user.account_id));
let webauthn = webauthn_for_auth(&auth_public_config(&api.config)).unwrap();
let (mut options, state) = webauthn
.start_passkey_registration(
random_handle,
&user.handle,
&user.display_name,
None,
)
.await;
assert_eq!(logged_in, expected);
}
let (cookie, _) = f
.login_session_with_user_handle(
"legacy",
&mut legacy_authenticator,
Some(legacy_user_handle.as_slice()),
)
.await;
let before_add = f.credentials(&expected);
let mut added = authenticator();
let options = f.options("addition", Some(&cookie), "").await;
let stable_user_handle = options.public_key.public_key.user.id.clone();
assert_eq!(
stable_user_handle.as_slice(),
passkey_user_handle(&user.account_id).as_bytes()
);
assert_ne!(stable_user_handle, legacy_user_handle);
let request = Fixture::credential(options, &mut added);
let (status, _, body) = f
.post(
"/api/auth/passkeys/addition/complete",
Some(&cookie),
Some(ORIGIN_URL),
request,
)
.await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(
serde_json::from_value::<AuthUserResponse>(body)
.unwrap()
.user,
expected
);
assert_eq!(
.unwrap();
if resident {
let selection = options.public_key.authenticator_selection.as_mut().unwrap();
selection.require_resident_key = true;
selection.resident_key =
Some(webauthn_rs_proto::ResidentKeyRequirement::Required);
}
let legacy_user_handle = options.public_key.user.id.clone();
assert_eq!(legacy_user_handle.as_slice(), random_handle.as_bytes());
let mut legacy_authenticator = authenticator();
let credential = legacy_authenticator
.do_registration(Url::parse(ORIGIN_URL).unwrap(), options)
.unwrap();
let passkey = webauthn
.finish_passkey_registration(&credential, &state)
.unwrap();
let record = PasskeyCredentialRecord {
credential_id: passkey_credential_id(&passkey).unwrap(),
user_id: user.user_id.clone(),
public_key_cose: serde_json::to_string(&passkey).unwrap(),
transports_json: None,
sign_count: 0,
created_at: crate::auth::now_rfc3339(),
last_used_at: None,
};
f.api
.store
.get_passkey_credential(&record.credential_id)
.unwrap()
.as_ref(),
Some(&before_add[0])
);
for returned_handle in [None, Some(legacy_user_handle.as_slice())] {
let (_, logged_in) = f
.register_verified_passkey(
&PasskeyRegistrationTarget::NewAccount { user: user.clone() },
&record,
)
.unwrap();
let expected = user_response(user.clone());
if resident {
let normal = f.discoverable_request(&mut legacy_authenticator).await;
rejected(
f.post(
"/api/auth/passkeys/login/complete",
None,
Some(ORIGIN_URL),
serde_json::to_value(normal).unwrap(),
)
.await,
StatusCode::BAD_REQUEST,
"passkey_user_mismatch",
);
} else {
let normal = f.discoverable_options().await;
assert!(
legacy_authenticator
.do_authentication(Url::parse(ORIGIN_URL).unwrap(), normal.public_key)
.is_err(),
"a non-resident key cannot be discovered without an allow list"
);
}
for returned_handle in [None, Some(legacy_user_handle.as_slice())] {
let (_, logged_in) = f
.login_session_with_user_handle(
"legacy",
&mut legacy_authenticator,
returned_handle,
)
.await;
assert_eq!(logged_in, expected);
}
let (cookie, _) = f
.login_session_with_user_handle(
"legacy",
&mut legacy_authenticator,
returned_handle,
Some(legacy_user_handle.as_slice()),
)
.await;
assert_eq!(logged_in, expected);
}
for returned_handle in [None, Some(stable_user_handle.as_slice())] {
let (_, logged_in) = f
.login_session_with_user_handle("legacy", &mut added, returned_handle)
let before_add = f.credentials(&expected);
let mut added = authenticator();
let options = f.options("addition", Some(&cookie), "").await;
let stable_user_handle = options.public_key.public_key.user.id.clone();
assert_eq!(
stable_user_handle.as_slice(),
passkey_user_handle(&user.account_id).as_bytes()
);
assert_ne!(stable_user_handle, legacy_user_handle);
let request = Fixture::credential(options, &mut added);
let (status, _, body) = f
.post(
"/api/auth/passkeys/addition/complete",
Some(&cookie),
Some(ORIGIN_URL),
request,
)
.await;
assert_eq!(logged_in, expected);
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(
serde_json::from_value::<AuthUserResponse>(body)
.unwrap()
.user,
expected
);
assert_eq!(
f.api
.store
.get_passkey_credential(&record.credential_id)
.unwrap()
.as_ref(),
Some(&before_add[0])
);
for returned_handle in [None, Some(legacy_user_handle.as_slice())] {
let (_, logged_in) = f
.login_session_with_user_handle(
"legacy",
&mut legacy_authenticator,
returned_handle,
)
.await;
assert_eq!(logged_in, expected);
}
for returned_handle in [None, Some(stable_user_handle.as_slice())] {
let (_, logged_in) = f
.login_session_with_user_handle("legacy", &mut added, returned_handle)
.await;
assert_eq!(logged_in, expected);
}
assert_eq!(f.discoverable_login(&mut added).await, expected);
}
}
}
#[tokio::test]
async fn generated_passkey_routes_complete_registration_login_and_logout_cookie_lifecycle() {
use webauthn_authenticator_rs::{WebauthnAuthenticator, softpasskey::SoftPasskey};
mod resident_model {
include!("../../../tests/e2e/src/resident_authenticator.rs");
}
use resident_model::ResidentAuthenticator;
let workspace = tempfile::tempdir().unwrap();
let mut api = test_api(workspace.path()).await;
@@ -53182,7 +53346,7 @@ mod tests {
let origin = origin.clone();
let cookie_name = cookie_name.clone();
let app = generated_auth_contract_router(ServerApiContractService::Workspace(api));
let mut authenticator = WebauthnAuthenticator::new(SoftPasskey::new(true));
let mut authenticator = ResidentAuthenticator::new(true);
let authenticator_origin = url::Url::parse(&origin).unwrap();
let registration_options_response = app
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,107 @@
//! Authenticated Subject-body access to durable Memory/Journal links.
//!
//! The current singleton grants access; identifiers in the request do not. Saved
//! Journal reads deliberately use a reader context, not a source Session, Runtime
//! lookup or restore. Domain storage owns link matching, persistence and paging.
use super::*;
use subjektiv::journal::memory_links::{MemoryJournalRequest, MemoryJournalResponse};
use subjektiv::journal::{JournalError, JournalHostContext};
fn domain(error: JournalError) -> Error {
match error {
JournalError::PermissionDenied(message) => Error::WorkspacePermissionDenied(message),
JournalError::Conflict(message) => Error::RepositoryConflict(message),
JournalError::InvalidInput(message) => Error::InvalidInput(message),
JournalError::NotFound(id) => {
Error::InvalidInput(format!("Journal record not found: {id}"))
}
JournalError::Subject(error) => subjektiv_store_error(error),
other => Error::Store(other.to_string()),
}
}
pub(super) async fn route(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
source: Option<axum::Extension<crate::worker_source::VerifiedRuntimeRequestSource>>,
Json(request): Json<MemoryJournalRequest>,
) -> ApiResult<Json<MemoryJournalResponse>> {
let source = source
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Memory Journal requires signed Runtime request identity".into(),
)
})?
.0;
let worker_id = source.worker_id.ok_or_else(|| {
Error::WorkspacePermissionDenied("Memory Journal requires bound Worker identity".into())
})?;
let worker = RuntimeWorkerRef::new(source.runtime_id, worker_id);
// This is entirely store-owned, including callbacks while a Host parent waits
// on Runtime. Do not put it behind Host or consult live Runtime/config state.
api.store
.db_executor()
.run(move || {
validate_workspace_scope(&api, &path.workspace_id)?;
let projection = api
.store
.worker_registry_projection(&path.workspace_id, &worker)?
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Memory Journal Worker registration missing".into(),
)
})?;
// No Job grant implicitly expands to all of a Subject's saved Journal.
if projection.job.is_some() {
return Err(Error::WorkspacePermissionDenied(
"Memory Journal is not authorized for Job Workers".into(),
)
.into());
}
let lease = api
.store
.require_current_worker_singleton_owner(&path.workspace_id, &worker)?
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Memory Journal requires the current Subject Worker".into(),
)
})?;
let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?;
require_subjektiv_worker_authority(
authority,
SubjektivWorkerAuthority::Subject,
"Memory Journal access",
)?;
let store = open_subjektiv_store(&api)?;
let host = JournalHostContext::validated_reader(&store, subject_id).map_err(domain)?;
store
.execute_memory_journal_with_source_presence(&host, request, |session| {
// Absence is known only when neither retained Worker registration nor an
// archive location remains. Runtime disconnection is not material deletion.
// These SQL lookups never fetch Session bytes or restore the old Worker.
let attribution = store.session_attribution(session).ok()??;
if attribution.subject_id != subject_id {
return None;
}
let target =
RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id);
let archive = api
.config_store
.worker_session_archive(&path.workspace_id, &target, session)
.ok()?;
let registered = api
.store
.get_worker_registry(&path.workspace_id, &target)
.ok()?;
if archive.is_none() && registered.is_none() {
Some(false)
} else {
None
}
})
.map(Json)
.map_err(domain)
.map_err(ApiError::from)
})
.await
}
@@ -110,7 +110,7 @@ async fn subject_callbacks_finish_before_waiting_host_parents_release_capacity()
));
}
// Resident memory is not an allowed deletion-drain callback. Preserve that
// admission fence; only the exact Session attribution route reenters below.
// admission fence; only exact Session attribution and startup read callbacks reenter below.
let mut deletion = Box::pin(gate.lock_deletion());
assert!(futures::poll!(&mut deletion).is_pending());
for (subject_id, worker_id, session_id) in &callbacks {
@@ -139,6 +139,27 @@ async fn subject_callbacks_finish_before_waiting_host_parents_release_capacity()
let stored = subjektiv.session_attribution(session_id).unwrap().unwrap();
assert_eq!(stored.worker_id, *worker_id);
assert_eq!(stored.subject_id, *subject_id);
let startup_path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/journal/startup");
let response = tokio::time::timeout(
std::time::Duration::from_secs(2),
app.clone().oneshot(runtime_source_request(
&identity,
Some(worker_id),
"GET",
&startup_path,
Vec::new(),
)),
)
.await
.expect("startup snapshot must not wait for Runtime's occupied Host parents")
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let snapshot: subjektiv::journal::startup::JournalStartupSnapshot =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(snapshot.subject_id, *subject_id);
assert!(snapshot.journals.is_empty());
assert!(!snapshot.omitted);
}
// Neither actual parent nor its admission lease was released to make these
// callbacks succeed. Only durable attribution completion releases them.
@@ -297,7 +318,7 @@ async fn only_exact_authenticated_callbacks_can_reenter_deletion_drain() {
}
#[tokio::test]
async fn rejected_create_releases_reservation_and_key_without_cleanup_errors() {
async fn rejected_create_releases_key_but_retains_subject_occupancy_without_release_proof() {
let workspace = tempfile::tempdir().unwrap();
let api = test_api(workspace.path()).await;
let memory = api
@@ -349,11 +370,13 @@ async fn rejected_create_releases_reservation_and_key_without_cleanup_errors() {
.unwrap()
.is_none()
);
assert!(
assert_eq!(
api.store
.current_worker_singleton_owner(TEST_WORKSPACE_ID, "subjektiv:rejected-create")
.unwrap()
.is_none()
.expect("failed creation is not explicit Subject release authority")
.worker,
worker,
);
}
}
@@ -0,0 +1,623 @@
//! Connected Backend → production Runtime factory → real Controller → Journal → signed
//! Host startup reader → persisted first input. Only the external model stream is scripted.
//! Browser authentication is outside this regression: creation enters the real scoped handler.
use super::*;
use agen::llm_client::{
LlmClient, Request,
error::ClientError,
event::{Event, StopReason},
types::Item,
};
use serde_json::{Value, json};
use std::sync::{Arc, Mutex};
#[derive(Clone, Default)]
struct ConnectedJournalModel {
dialogs: Arc<Mutex<Vec<Request>>>,
}
impl ConnectedJournalModel {
fn journal_response(request: &Request) -> Vec<Event> {
// Runtime Controllers deliberately retain alias-routed core schemas for their
// lifetime even with zero attachments. Verify invocation authority, not schema
// absence (standalone Internal Jobs use a different composition).
for (id, tool, input) in [
(
"denied-bash",
"Bash",
json!({"command":"printf JOURNAL_MUST_NOT_EXECUTE","cwd":"."}),
),
(
"denied-read",
"Read",
json!({"file_path":"journal-must-not-be-readable"}),
),
] {
if let Some(error) = request.items.iter().find_map(|item| match item {
Item::ToolResult {
call_id, is_error, ..
} if call_id == id => Some(*is_error),
_ => None,
}) {
assert!(
error,
"Journal Job unexpectedly gained {tool} Workdir authority"
);
} else {
return vec![
Event::tool_use_start(0, id, tool),
Event::tool_input_delta(0, input.to_string()),
Event::tool_use_stop(0),
];
}
}
let result = |id: &str| -> Option<Value> {
request.items.iter().find_map(|item| match item {
Item::ToolResult {
call_id,
content,
is_error,
summary,
..
} if call_id == id => {
assert!(!is_error, "{id}: {summary}; {content:?}");
Some(
content
.as_deref()
.map(|text| serde_json::from_str(text).unwrap())
.unwrap_or(Value::Null),
)
}
_ => None,
})
};
if result("submit").is_some() {
return vec![
Event::text_block_start(0),
Event::text_delta(0, "saved"),
Event::text_block_stop(0, Some(StopReason::EndTurn)),
];
}
let (tool, id, input) = if result("checkpoint").is_some() {
("SubmitBackendJobResult", "submit", json!({"result":{}}))
} else if let Some(position) = result("position") {
let state = result("state").unwrap();
(
"CheckpointJournal",
"checkpoint",
json!({
"request_id":format!("checkpoint:{}",state["value"]["generation_id"].as_str().unwrap()),
"expected_state":position["value"]["state_token"],
"action":{"action":"checkpoint","record_operation_ids":[format!("record:{}",state["value"]["generation_id"].as_str().unwrap())]},
"covered_ranges":state["value"]["selected_ranges"],
}),
)
} else if let Some(record) = result("record") {
let state = result("state").unwrap();
(
"UpdateJournalPosition",
"position",
json!({
"request_id":format!("position:{}",state["value"]["generation_id"].as_str().unwrap()),
"expected_state":record["value"]["state_token"],
"action":{"action":"update_position","current_position":"CONNECTED_HANDOFF: partially implemented; outcome unconfirmed"},
"covered_ranges":[],
}),
)
} else if let Some(state) = result("state") {
let state = &state["value"];
// Deliver the actual frozen Runtime source through authenticated Job tools, rather
// than injecting a fixture Journal or claiming unseen source was processed.
for (index, range) in state["selected_ranges"]
.as_array()
.unwrap()
.iter()
.enumerate()
{
let mut offset = range["bytes"]["start"].as_u64().unwrap();
let mut token = Value::Null;
loop {
let id = format!("source-{index}-{offset}");
if let Some(page) = result(&id) {
let page = &page["value"];
if let Some(next) = page["next_byte"].as_u64() {
assert!(next > offset);
offset = next;
token = page["content_token"].clone();
} else {
break;
}
} else {
assert!(
request
.tools
.iter()
.any(|tool| tool.name == "ReadJournalText")
);
return vec![Event::tool_use_start(0, &id, "ReadJournalText"), Event::tool_input_delta(0, json!({
"target":{"target":"source","pointer":range["source"]["pointer"]},
"offset":offset,"limit":MAX_JOURNAL_READ_BYTES,"content_token":token,
}).to_string()), Event::tool_use_stop(0)];
}
}
}
let (tool, action) = if state["current_open"].is_null() {
(
"CreateJournalEntry",
json!({"action":"create","title":"Connected handoff","body":"CONNECTED_HANDOFF: proposed A, user chose B; only partial work performed.","ongoing":"Remaining work; do not claim success","state":"open","source_end":null,"correction_refs":[]}),
)
} else {
(
"RewriteJournalEntry",
json!({"action":"rewrite","entry_id":state["current_open"]["id"],"title":"Connected handoff","body":"CONNECTED_HANDOFF: B selected; partial work remains.","ongoing":"Remaining work; result unknown"}),
)
};
(
tool,
"record",
json!({
"request_id":format!("record:{}",state["generation_id"].as_str().unwrap()),
"expected_state":state["current_journal"]["state_token"],
"action":action,"covered_ranges":state["selected_ranges"],
}),
)
} else {
("ReadJournalState", "state", json!({}))
};
assert!(
request
.tools
.iter()
.any(|definition| definition.name == tool),
"missing {tool}"
);
for forbidden in ["MemoryApplyCandidate", "SubjektivSessionRead"] {
assert!(
!request
.tools
.iter()
.any(|definition| definition.name == forbidden),
"Journal exposed forbidden {forbidden}: {:?}",
request
.tools
.iter()
.map(|tool| &tool.name)
.collect::<Vec<_>>()
);
}
vec![
Event::tool_use_start(0, id, tool),
Event::tool_input_delta(0, input.to_string()),
Event::tool_use_stop(0),
]
}
}
#[async_trait::async_trait]
impl LlmClient for ConnectedJournalModel {
async fn stream(
&self,
request: Request,
) -> std::result::Result<
std::pin::Pin<
Box<dyn futures::Stream<Item = std::result::Result<Event, ClientError>> + Send>,
>,
ClientError,
> {
let events = if request
.tools
.iter()
.any(|tool| tool.name == "ReadJournalState")
{
Self::journal_response(&request)
} else if request.tools.iter().any(|tool| {
matches!(
tool.name.as_str(),
"SubmitBackendJobResult" | "SubmitMemorySurface"
)
}) {
// The real companion also requests an empty Memory consolidation Job. Keep that
// lifecycle real without counting its model requests as main-body dialog requests.
if request.items.iter().any(|item| matches!(item, Item::ToolResult { call_id, is_error, .. } if call_id == "empty-memory" && !is_error)) {
vec![Event::text_block_start(0), Event::text_delta(0, "empty Memory acknowledged"), Event::text_block_stop(0, Some(StopReason::EndTurn))]
} else {
let (tool, input) = if request.tools.iter().any(|tool| tool.name == "SubmitMemorySurface") {
("SubmitMemorySurface", json!({"points":[]}))
} else {
let input: Value = request.items.iter().filter_map(Item::as_text).find_map(|text| text.lines().find_map(|line| line.strip_prefix("input_json: "))).map(|input| serde_json::from_str(input).unwrap()).expect("Backend Memory Job envelope");
assert_eq!(input["candidate_ids"], json!([]));
("SubmitBackendJobResult", json!({"result":{"subject_id":input["subject_id"],"candidate_ids":[]}}))
};
vec![Event::tool_use_start(0, "empty-memory", tool), Event::tool_input_delta(0, input.to_string()), Event::tool_use_stop(0)]
}
} else {
self.dialogs.lock().unwrap().push(request);
vec![
Event::text_block_start(0),
Event::text_delta(
0,
"Partial B implementation; remaining work is not confirmed.",
),
Event::text_block_stop(0, Some(StopReason::EndTurn)),
]
};
Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
}
fn clone_boxed(&self) -> Box<dyn LlmClient> {
Box::new(self.clone())
}
}
async fn connected_create(api: &WorkspaceApi, subject: &str, name: &str) -> RuntimeWorkerRef {
let Json(created) = scoped_create_workspace_worker(
State(api.clone()),
AxumPath(ScopedWorkspacePath { workspace_id: TEST_WORKSPACE_ID.into() }),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(), display_name: name.into(),
singleton_key: None, profile: Some("builtin:companion".into()), ticket_assignment: None,
initial_submit: vec![protocol::Segment::Text { content: "User chose B rather than A. Continue the partial work without assuming success.".into() }],
workdir_attachments: vec![],
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { subject_id: subject.into() }),
}, control_operation_id: None,
}),
).await.unwrap();
RuntimeWorkerRef::new(created.runtime_id, created.worker_id)
}
async fn connected_wait_for_recording(
api: &WorkspaceApi,
store: &crate::subjektiv::SubjektivStore,
attr: &crate::subjektiv::SubjectSessionAttribution,
) {
let host = JournalHostContext::validated_session(store, attr.clone()).unwrap();
tokio::time::timeout(std::time::Duration::from_secs(30), async {
let mut tick = tokio::time::interval(std::time::Duration::from_millis(25));
loop {
tick.tick().await;
let jobs = api
.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &attr.session_id, None, 100)
.unwrap();
let generation_closed =
store
.journal_for_session(&host)
.unwrap()
.is_some_and(|journal| {
store
.active_journal_generation(&host, &journal)
.unwrap()
.is_none()
});
if !jobs.is_empty()
&& generation_closed
&& jobs
.iter()
.all(|job| job.state == BackendJobState::Completed)
&& !store.journal_retention_required(&host).unwrap()
&& api
.runtime
.worker(&RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id))
.unwrap()
.worker_state
.is_some_and(|state| state.catalog_status() == protocol::WorkerStatus::Idle)
{
break;
}
}
})
.await
.unwrap_or_else(|_| {
panic!(
"real Controller recording did not complete: {:?}",
api.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &attr.session_id, None, 100)
.unwrap()
)
});
}
#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() {
let dir = tempfile::tempdir().unwrap();
init_clean_git_workspace(dir.path());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let mut config = test_server_config(dir.path());
config.backend_base_url = Some(format!("http://{}", listener.local_addr().unwrap()));
let control = Arc::new(test_control_store(&config));
seed_test_registered_workspace(control.as_ref(), &config)
.await
.unwrap();
let broker = BackendResourceBroker::default();
let dispatcher = Arc::new(
crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new(
config.clone(),
control.clone(),
),
);
let model = ConnectedJournalModel::default();
let model_factory = model.clone();
let audience = embedded_runtime_request_audience(&config).unwrap();
let execution = worker_runtime::worker_backend::WorkerRuntimeExecutionBackend::new(
worker_runtime::worker_backend::ProfileRuntimeWorkerFactory::new(
config.workspace_execution_root.clone(),
)
.with_embedded_worker_mutation_dispatcher(EMBEDDED_WORKER_RUNTIME_ID, dispatcher.clone())
.with_workspace_request_client(
worker_runtime::workspace_request::RuntimeWorkspaceRequestClient::new(
TEST_WORKSPACE_ID,
audience.clone(),
EMBEDDED_WORKER_RUNTIME_ID,
)
.with_runtime_request_source(&EMBEDDED_RUNTIME_REQUEST_IDENTITY, audience),
)
.with_runtime_store_dir(config.embedded_runtime_store_root.clone())
.with_resource_client(Arc::new(broker.clone()))
.with_spawn_model_client_factory(move || Box::new(model_factory.clone())),
)
.unwrap();
let api = WorkspaceApi::new_with_execution_backend_and_broker(
config,
control,
Arc::new(execution),
broker,
HostExecutors::default(),
Some(dispatcher),
)
.await
.unwrap();
seed_test_delivery_binding(&api);
let requests = Arc::new(Mutex::new(Vec::new()));
let observed = requests.clone();
let app = build_router(api.clone()).layer(axum::middleware::from_fn(
move |request: axum::extract::Request, next: axum::middleware::Next| {
let observed = observed.clone();
async move {
let path = request.uri().path().to_string();
let response = next.run(request).await;
observed.lock().unwrap().push((path, response.status()));
response
}
},
));
let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(crate::subjektiv::SubjectRole::new("connected-handoff").unwrap())
.unwrap();
let old = connected_create(&api, &subject.id, "connected-old-body").await;
let old_attr = store
.subject_session_attributions(&subject.id)
.unwrap()
.into_iter()
.find(|attr| attr.worker_id == old.worker_id)
.unwrap();
connected_wait_for_recording(&api, &store, &old_attr).await;
let old_host = JournalHostContext::validated_session(&store, old_attr.clone()).unwrap();
let old_binding = JournalWriterBinding {
subject_id: subject.id.clone(),
runtime_id: old.runtime_id.clone(),
worker_id: old.worker_id.clone(),
binding_id: old_host.writer_binding_id().unwrap().into(),
};
// Inject a durable result-acceptance gap after real Journal completion. This must
// reject before writer revocation and confirm cancellation of the actual Controller
// guard; merely returning Err would strand the body and its execution-admission fence.
let recorded = api
.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &old_attr.session_id, None, 100)
.unwrap();
let uncertain_job = &recorded[0].request.job_id;
let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
let saved_result: (String, String) = conn.query_row(
"SELECT result_json,result_digest FROM backend_jobs WHERE workspace_id=?1 AND job_id=?2",
rusqlite::params![TEST_WORKSPACE_ID, uncertain_job], |r| Ok((r.get(0)?, r.get(1)?)),
).unwrap();
conn.execute(
"UPDATE backend_jobs SET state='unknown',result_json=NULL,result_digest=NULL WHERE workspace_id=?1 AND job_id=?2",
rusqlite::params![TEST_WORKSPACE_ID, uncertain_job],
)
.unwrap();
let incomplete = json!({
"worker":old,"binding_id":old_binding.binding_id,"operation_id":"connected-incomplete-release",
});
let error = subjektiv_journal::release_subject(
&api,
serde_json::from_value(incomplete.clone()).unwrap(),
)
.unwrap_err();
assert!(error.to_string().contains("Unknown"), "{error}");
assert!(store.journal_writer_active(&old_binding).unwrap());
assert!(
!api.config_store
.subject_release_pending(TEST_WORKSPACE_ID, &old)
.unwrap()
);
let cancelled: bool = conn.query_row("SELECT cancelled FROM subject_worker_releases WHERE workspace_id=?1 AND operation_id='connected-incomplete-release'", [TEST_WORKSPACE_ID], |r| r.get(0)).unwrap();
assert!(
cancelled,
"checked pre-revoke refusal needs a terminal cancellation receipt"
);
assert!(
subjektiv_journal::release_subject(&api, serde_json::from_value(incomplete).unwrap())
.is_err(),
"cancelled release identity must not rearm"
);
conn.execute(
"UPDATE backend_jobs SET state='completed',result_json=?3,result_digest=?4 WHERE workspace_id=?1 AND job_id=?2",
rusqlite::params![TEST_WORKSPACE_ID, uncertain_job, saved_result.0, saved_result.1],
)
.unwrap();
// The new release below must reacquire the genuine Controller guard. It cannot
// succeed if the failed pre-revoke operation left the body frozen.
let release = subjektiv_journal::release_subject(&api, serde_json::from_value(json!({
"worker":old,"binding_id":old_binding.binding_id,"operation_id":"connected-normal-release",
})).unwrap()).unwrap();
assert_eq!(
release.state,
runtime_api::SubjectReleaseState::Disconnected
);
assert!(!store.journal_writer_active(&old_binding).unwrap());
let key = crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap();
assert!(
api.store
.current_worker_singleton_owner(TEST_WORKSPACE_ID, &key)
.unwrap()
.is_none()
);
let stopped = api
.runtime
.stop_worker(&old, WorkerLifecycleRequest::default())
.unwrap();
assert_eq!(stopped.state, InternalWorkerOperationState::Accepted);
let removed = WorkerRemovalService::new(&api)
.execute_target_removal(
api.runtime.as_ref(),
&old,
"connected regression removes prior body",
None,
)
.await
.unwrap();
assert_eq!(removed.status, StatusCode::OK.as_u16(), "{}", removed.body);
assert!(api.runtime.worker(&old).is_err());
let original_root = api
.config
.embedded_runtime_store_root
.join("workers")
.join(&old.worker_id)
.join("session");
assert!(
!original_root.exists(),
"prior Worker original Session must actually be removed before startup"
);
// Purge retained original archive through Runtime, not fixture filesystem deletion.
// Journal text is independent and remains readable for new-body startup.
if let Some(archive) = api
.config_store
.worker_tombstone(TEST_WORKSPACE_ID, &old)
.unwrap()
.and_then(|t| t.archive_id)
{
api.runtime
.session_material_erase(
&old.runtime_id,
runtime_api::SessionMaterialEraseRequest {
workspace_id: TEST_WORKSPACE_ID.into(),
source: runtime_api::SessionPublicSource::Archived {
archive_id: archive.clone(),
},
expected_session_id: old_attr.session_id.clone(),
},
)
.unwrap();
assert!(
!api.config
.embedded_runtime_store_root
.join("archives/workers")
.join(archive)
.join("session")
.exists()
);
}
let next = connected_create(&api, &subject.id, "connected-next-body").await;
assert_ne!(next, old);
let next_attr = store
.subject_session_attributions(&subject.id)
.unwrap()
.into_iter()
.find(|attr| attr.worker_id == next.worker_id)
.unwrap();
assert_ne!(next_attr.session_id, old_attr.session_id);
connected_wait_for_recording(&api, &store, &next_attr).await;
// Read actual Session files produced by the real Controller; no fixture Session append,
// startup-item injection, fake disconnect or store-only simulation is used in this test.
use session_store::Store as _;
let session = session_store::WorkerSessionStore::open_read_only(
api.config
.embedded_runtime_store_root
.join("workers")
.join(&next.worker_id)
.join("session"),
)
.unwrap();
let session_id = next_attr.session_id.parse().unwrap();
let segments = session.list_segments(session_id).unwrap();
assert_eq!(segments.len(), 1);
let entries = session.read_all_read_only(session_id, segments[0]).unwrap();
let startup: Vec<_> = entries
.iter()
.enumerate()
.filter_map(|(index, log)| match log {
session_store::LogEntry::AnnotatedSystemItem { entry, .. } => match &entry.item {
session_store::SystemItem::JournalStartup {
snapshot,
snapshot_sha256,
body,
prompt_provenance,
} => Some((index, snapshot, snapshot_sha256, body, prompt_provenance)),
_ => None,
},
_ => None,
})
.collect();
assert_eq!(startup.len(), 1, "startup must be durably committed once");
let (startup_index, snapshot, digest, body, provenance) = startup[0];
let first_input = entries
.iter()
.position(|entry| matches!(entry, session_store::LogEntry::AnnotatedUserInput { .. }))
.expect("new body persisted its initial human input");
assert!(
startup_index < first_input,
"startup reference must be persisted before the first human input/model request"
);
assert_eq!(snapshot["subject_id"], subject.id);
assert_eq!(
snapshot["journals"][0]["source_session_id"], old_attr.session_id,
"Host-selected writer predecessor must lead the snapshot"
);
assert!(body.contains("CONNECTED_HANDOFF"));
assert_eq!(
provenance.logical_name,
"internal.subjektiv_journal_startup"
);
assert_eq!(
*digest,
Sha256::digest(serde_json::to_vec(snapshot).unwrap())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
);
let dialogs = model.dialogs.lock().unwrap();
assert!(dialogs.len() >= 2);
assert!(
dialogs[1]
.items
.iter()
.filter_map(Item::as_text)
.any(|text| text.contains(body)),
"first new-body model request must contain the same persisted startup body"
);
drop(dialogs);
let requests = requests.lock().unwrap();
assert_eq!(
requests
.iter()
.filter(
|(path, status)| path.ends_with("/subjektiv/journal/startup")
&& *status == StatusCode::OK
)
.count(),
2,
"both bodies must use the authenticated HTTP Host startup reader"
);
assert!(
requests.iter().any(
|(path, status)| path.ends_with("/subjektiv/sessions") && *status == StatusCode::OK
)
);
drop(requests);
server.abort();
}
@@ -8,6 +8,19 @@ use sha2::{Digest, Sha256};
use subjektiv::journal::transport::*;
use subjektiv::journal::*;
#[path = "subjektiv_journal_connected_tests.rs"]
mod connected;
fn bind_fixture_session(
store: &crate::subjektiv::SubjektivStore,
attr: crate::subjektiv::SubjectSessionAttribution,
) -> JournalHostContext {
let binding = store
.claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
.unwrap();
store.bind_journal_session(&binding, attr).unwrap()
}
#[tokio::test]
async fn journal_job_reserves_before_binding_and_retains_partial_receipts_on_cancel() {
let Fixture {
@@ -195,7 +208,7 @@ impl Fixture {
.unwrap(),
)
.unwrap();
let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap();
let host = bind_fixture_session(&store, attr.clone());
let text = "original public input";
let range = JournalSourceRange {
source: JournalSource {
@@ -228,6 +241,7 @@ impl Fixture {
profile: "builtin:subjektiv-journal-recording".into(),
grants: BackendJobGrants {
subjektiv_journal: Some(SubjektivJournalGrant {
binding_id: host.writer_binding_id().unwrap().into(),
subject_id: subject.id,
session_id: attr.session_id,
journal_id: id.clone(),
@@ -288,13 +302,14 @@ async fn journal_completion_replays_only_saved_checkpoint_and_fences_late_operat
panic!("receipt required")
};
assert!(subjektiv_journal::finish(&f.api, worker).is_err());
let position = save_fixture_position(&f.api, worker, receipt.state_token);
subjektiv_journal::execute(
&f.api,
worker,
JournalJobRequest::Execute {
operation: JournalOperation {
request_id: "checkpoint".into(),
expected_state: receipt.state_token,
expected_state: position,
action: JournalAction::Checkpoint {
record_operation_ids: vec!["no-change".into()],
},
@@ -332,6 +347,69 @@ async fn journal_completion_replays_only_saved_checkpoint_and_fences_late_operat
);
assert!(subjektiv_journal::execute(&f.api, worker, JournalJobRequest::ReadState).is_err());
assert!(!f.store.journal_retention_required(&f.host).unwrap());
// Store authority boundary: Host Job-state gating is separate from revocation itself.
let grant = f.request.grants.subjektiv_journal.as_ref().unwrap();
let attr = f
.store
.session_attribution(&grant.session_id)
.unwrap()
.unwrap();
let binding = JournalWriterBinding {
subject_id: attr.subject_id.clone(),
runtime_id: attr.runtime_id.clone(),
worker_id: attr.worker_id.clone(),
binding_id: grant.binding_id.clone(),
};
f.store.release_journal_writer(&binding).unwrap();
let next = f
.store
.claim_journal_writer(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
.unwrap();
assert_ne!(next.binding_id, binding.binding_id);
assert!(f.store.bind_journal_session(&next, attr).is_err());
let mut stale = dispatched.clone();
stale
.job
.request
.grants
.subjektiv_journal
.as_mut()
.unwrap()
.binding_id = next.binding_id;
assert!(subjektiv_journal::prepare(&f.api, &stale).is_err());
assert!(
f.store
.query_journal_operation(&f.host, &f.id, "no-change")
.unwrap()
.is_some()
);
// Exact saved completion is still recoverable; it is not a new write grant.
subjektiv_journal::validate_result(&f.api, &f.request, &dispatched.attempt, &value).unwrap();
}
fn save_fixture_position(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
expected_state: String,
) -> String {
let JournalJobResponse::Saved(receipt) = subjektiv_journal::execute(
api,
worker,
JournalJobRequest::Execute {
operation: JournalOperation {
request_id: "position".into(),
expected_state,
action: JournalAction::UpdatePosition {
current_position: "No activity change; work remains pending.".into(),
},
covered_ranges: vec![],
},
},
)
.unwrap() else {
panic!("saved position receipt required")
};
receipt.state_token
}
fn deliver_fixture_source(
@@ -483,7 +561,7 @@ impl RuntimeFixture {
.unwrap(),
)
.unwrap();
let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap();
let host = bind_fixture_session(&store, attr.clone());
let src = runtime_api::SessionPublicSource::Retained {
worker_id: worker_runtime::identity::WorkerId::parse(&source.worker_id).unwrap(),
};
@@ -543,13 +621,15 @@ impl RuntimeFixture {
.accept_journal_sources(&host, "source", &[range.clone()])
.unwrap();
let request = subjektiv_journal::recording_request(
&host,
attr,
&id,
vec![subjektiv::journal::recording::JournalMaterial {
range: range.clone(),
text: SOURCE_SENTINEL.into(),
}],
);
)
.unwrap();
assert!(
!serde_json::to_string(&request)
.unwrap()
@@ -650,13 +730,14 @@ impl RuntimeFixture {
.unwrap() else {
panic!("receipt")
};
let position = save_fixture_position(&self.api, worker, receipt.state_token);
subjektiv_journal::execute(
&self.api,
worker,
JournalJobRequest::Execute {
operation: JournalOperation {
request_id: "checkpoint".into(),
expected_state: receipt.state_token,
expected_state: position,
action: JournalAction::Checkpoint {
record_operation_ids: vec!["record".into()],
},
@@ -1661,7 +1742,7 @@ impl SubjectRemovalFixture {
.unwrap(),
)
.unwrap();
let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap();
let host = bind_fixture_session(&store, attr.clone());
seed_worker_control_grant(&api, &controller, &target, "subject-removal-fixture");
api.runtime
.stop_worker(
@@ -1696,6 +1777,43 @@ impl SubjectRemovalFixture {
registry.profile = Some("builtin:default".into());
self.api.store.upsert_worker_registry(&registry).unwrap();
}
/// Seed the historical release boundary for retention-recovery component tests.
/// This deterministic stopped-Worker fixture has no Controller and does NOT
/// prove disconnect. The connected test below exercises actual release proof.
fn seed_completed_release_for_archive_recovery(&self) {
let lease = self
.api
.store
.require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &self.target)
.unwrap()
.unwrap();
let binding = JournalWriterBinding {
subject_id: self.attr.subject_id.clone(),
runtime_id: self.target.runtime_id.clone(),
worker_id: self.target.worker_id.clone(),
binding_id: self.host.writer_binding_id().unwrap().into(),
};
let operation = "historical-release-retention-fixture";
self.api
.config_store
.reserve_subject_release(TEST_WORKSPACE_ID, &lease, &binding.binding_id, operation)
.unwrap();
self.store.release_journal_writer(&binding).unwrap();
self.api
.config_store
.confirm_subject_release_proof(
TEST_WORKSPACE_ID,
&lease,
&binding.binding_id,
operation,
false,
)
.unwrap();
self.api
.config_store
.release_subject_singleton(TEST_WORKSPACE_ID, &lease, &binding.binding_id, operation)
.unwrap();
}
fn remove(&self) -> worker::WorkspaceResponse {
let key = self
.api
@@ -1714,6 +1832,37 @@ impl SubjectRemovalFixture {
}
}
#[tokio::test]
async fn subject_release_never_substitutes_catalog_idle_or_stopped_for_controller_reservation() {
let f = SubjectRemovalFixture::new().await;
let input = serde_json::from_value(serde_json::json!({
"worker":f.target,"binding_id":f.host.writer_binding_id().unwrap(),"operation_id":"actual-controller-required",
})).unwrap();
let error = subjektiv_journal::release_subject(&f.api, input).unwrap_err();
assert!(
error.to_string().contains("actual") || error.to_string().contains("Controller"),
"{error}"
);
let key = crate::subjektiv::subject_worker_singleton_key(&f.attr.subject_id).unwrap();
assert_eq!(
f.api
.store
.current_worker_singleton_owner(TEST_WORKSPACE_ID, &key)
.unwrap()
.unwrap()
.worker,
f.target
);
let binding = JournalWriterBinding {
subject_id: f.attr.subject_id.clone(),
runtime_id: f.target.runtime_id.clone(),
worker_id: f.target.worker_id.clone(),
binding_id: f.host.writer_binding_id().unwrap().into(),
};
assert!(f.store.journal_writer_active(&binding).unwrap());
assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
}
#[tokio::test]
async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a_job() {
let f = SubjectRemovalFixture::new().await;
@@ -1721,7 +1870,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a
let response = f.remove();
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
StatusCode::CONFLICT.as_u16(),
"{}",
response.body
);
@@ -1730,7 +1879,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a
.store
.get_worker_registry(TEST_WORKSPACE_ID, &f.target)
.unwrap()
.is_none()
.is_some()
);
assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
assert!(
@@ -1795,7 +1944,7 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() {
let response = f.remove();
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
StatusCode::CONFLICT.as_u16(),
"{}",
response.body
);
@@ -1804,62 +1953,25 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() {
.store
.get_worker_registry(TEST_WORKSPACE_ID, &f.target)
.unwrap()
.is_none()
.is_some()
);
let jobs = f
.api
.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
.unwrap();
assert_eq!(jobs.len(), 1);
assert_eq!(
jobs[0]
.request
.grants
.subjektiv_journal
.as_ref()
assert!(
f.api
.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
.unwrap()
.selected_ranges,
ranges[..1]
.is_empty()
);
assert_eq!(
f.store.journal_for_session(&f.host).unwrap().as_deref(),
Some(id.as_str())
);
assert!(f.store.journal_retention_required(&f.host).unwrap());
let archive = f
.api
.config_store
.worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id)
.unwrap()
.unwrap();
assert!(!session_root(f.dir.path(), &f.target).exists());
let source = runtime_api::SessionPublicSource::Archived {
archive_id: archive.archive_id,
};
let response = f
.api
.runtime
.session_material_catalog(
&f.target.runtime_id,
runtime_api::SessionMaterialCatalogRequest {
workspace_id: TEST_WORKSPACE_ID.into(),
source,
expected_session_id: f.attr.session_id.clone(),
expected_catalog_sha256: None,
offset: 0,
limit: 100,
},
)
.unwrap();
assert!(matches!(
response,
runtime_api::SessionMaterialCatalogAvailability::Page { .. }
));
assert!(session_root(f.dir.path(), &f.target).exists());
}
#[tokio::test]
async fn attributed_unmaterialized_subject_removal_needs_no_journal_job() {
async fn attributed_unmaterialized_subject_still_requires_actual_idle_release() {
let f = SubjectRemovalFixture::with_materialized_source(false).await;
let retention = source_retention(f.dir.path());
let session = f.attr.session_id.parse().unwrap();
@@ -1882,28 +1994,21 @@ async fn attributed_unmaterialized_subject_removal_needs_no_journal_job() {
.unwrap();
assert_eq!(retention.pins(session).unwrap().len(), 1);
let response = f.remove();
assert!(retention.pins(session).unwrap().is_empty());
assert_eq!(
response.status,
StatusCode::OK.as_u16(),
StatusCode::CONFLICT.as_u16(),
"{}",
response.body
);
assert_eq!(retention.pins(session).unwrap().len(), 1);
assert!(
f.api
.store
.get_worker_registry(TEST_WORKSPACE_ID, &f.target)
.unwrap()
.is_none()
.is_some()
);
assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
assert!(
f.api
.store
.list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
.unwrap()
.is_empty()
);
}
#[tokio::test]
@@ -2237,6 +2342,7 @@ async fn assert_erased_archive_plan_replans_without_republishing_sources(
) {
let mut f = SubjectRemovalFixture::new().await;
f.disable();
f.seed_completed_release_for_archive_recovery();
let policy = f
.api
.config_store
@@ -2331,6 +2437,7 @@ journal_recovery_cases! {
async fn assert_erased_completed_archive_receipt_is_audited_without_republishing(restart: bool) {
let mut f = SubjectRemovalFixture::new().await;
f.disable();
f.seed_completed_release_for_archive_recovery();
let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap();
// Runtime has committed source_removed, but Backend commit/response is lost.
conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'injected metadata commit interruption'); END;").unwrap();
@@ -2406,6 +2513,7 @@ journal_recovery_cases! {
async fn erased_archive_without_receipt_or_inventory_remains_unknown_without_replanning() {
let f = SubjectRemovalFixture::new().await;
f.disable();
f.seed_completed_release_for_archive_recovery();
let conn = rusqlite::Connection::open(&f.api.config.database_path).unwrap();
conn.execute_batch("CREATE TRIGGER interrupt_removal BEFORE UPDATE OF state ON worker_removal_operations WHEN NEW.state='executing' BEGIN SELECT RAISE(ABORT,'injected interruption'); END;").unwrap();
assert_ne!(f.remove().status, StatusCode::OK.as_u16());
@@ -2465,6 +2573,7 @@ async fn assert_erased_archive_retained_diagnostics_receipt_finalizes_only_after
) {
let mut f = SubjectRemovalFixture::new().await;
f.disable();
f.seed_completed_release_for_archive_recovery();
let policy = f
.api
.config_store
@@ -2662,3 +2771,36 @@ journal_recovery_cases! {
erased_archive_retained_diagnostics_reopen_finalizes_only_after_physical_ack => (true, false),
erased_archive_retained_diagnostics_interrupted_reopen_finalizes_only_after_physical_ack => (true, true),
}
#[tokio::test]
async fn legacy_journal_grant_preserves_old_json_and_only_recovers_original_current_body() {
let f = SubjectRemovalFixture::new().await;
let old = json!({"subject_id":f.attr.subject_id,"session_id":f.attr.session_id,"journal_id":"legacy-journal","selected_ranges":[]});
let wire = serde_json::to_string(&old).unwrap();
let grant: SubjektivJournalGrant = serde_json::from_str(&wire).unwrap();
assert!(grant.binding_id.is_empty());
assert_eq!(
serde_json::to_value(&grant).unwrap(),
old,
"old canonical request/digest must not acquire a new empty binding_id field"
);
assert_eq!(
serde_json::to_string(&serde_json::to_value(&grant).unwrap()).unwrap(),
wire
);
assert!(subjektiv_journal::host(&f.api, &f.store, &grant).is_ok());
let writer = JournalWriterBinding {
subject_id: f.attr.subject_id.clone(),
runtime_id: f.attr.runtime_id.clone(),
worker_id: f.attr.worker_id.clone(),
binding_id: f.host.writer_binding_id().unwrap().into(),
};
f.store.release_journal_writer(&writer).unwrap();
f.store
.claim_journal_writer(&f.attr.subject_id, &f.attr.runtime_id, "different-body")
.unwrap();
assert!(
subjektiv_journal::host(&f.api, &f.store, &grant).is_err(),
"legacy grant must never map to a later writer"
);
}
@@ -0,0 +1,731 @@
//! Connected ordinary library tests: production router, signed Runtime authentication,
//! singleton/Job authority and Subjektiv FeatureStorage/SQLite. Extraction output and
//! Host admission/generation are fixture inputs; staging/adoption and canonical Session
//! serialization are real. No Runtime execution, model, external config or network is used.
use super::*;
use subjektiv::journal::memory_links::*;
use subjektiv::journal::transport::{JournalJobRequest, JournalTextRead, JournalTextTarget};
use subjektiv::journal::*;
const RUNTIME: &str = "runtime-memory-journal";
const JOURNAL_BODY: &str =
"A was proposed; B was chosen after checking the constraint. Work remains.";
fn route_path() -> String {
format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/journal/memory")
}
async fn response(app: Router, request: Request<Body>) -> (StatusCode, Vec<u8>) {
let response = app.oneshot(request).await.unwrap();
let status = response.status();
let bytes = to_bytes(response.into_body(), 128 * 1024).await.unwrap();
(status, bytes.to_vec())
}
struct Fixture {
_dir: tempfile::TempDir,
api: WorkspaceApi,
identity: worker_runtime::auth::RuntimeIdentityMaterial,
store: crate::subjektiv::SubjektivStore,
host: JournalHostContext,
job: JournalJobContext,
worker: RuntimeWorkerRef,
source_worker: RuntimeWorkerRef,
source_root: PathBuf,
selector: MemoryJournalSelector,
journal_id: String,
entry_id: String,
ranges: Vec<JournalSourceRange>,
}
impl Fixture {
async fn new() -> Self {
use session_store::{LogEntry, Store as _, WorkerSessionStore};
let dir = tempfile::tempdir().unwrap();
let mut api = test_api(dir.path()).await;
let identity = worker_runtime::auth::RuntimeIdentityMaterial::generate(RUNTIME).unwrap();
configure_runtime_request_auth(&mut api, &identity, RUNTIME);
let store = open_subjektiv_store(&api).unwrap();
let subject = store
.create_subject(crate::subjektiv::SubjectRole::new("memory-journal").unwrap())
.unwrap();
let worker = Self::body_worker(&api, &subject.id);
let source_worker = RuntimeWorkerRef::new(RUNTIME, Uuid::new_v4().to_string());
seed_worker_source_member(&api, RUNTIME, &source_worker.worker_id);
let session = session_store::new_session_id();
let segment = Uuid::new_v4();
let source_root = dir.path().join("old-worker/session");
let history = [
"Choose B after checking the constraint",
"Correction: B still needs validation",
]
.into_iter()
.enumerate()
.map(|(index, text)| session_store::LoggedHistoryEntry {
item: session_store::LoggedItem::Message {
role: session_store::LoggedRole::User,
content: vec![session_store::LoggedContentPart::Text { text: text.into() }],
},
metadata: session_store::LoggedSessionHistoryMetadata {
entry_id: session_store::LoggedSessionHistoryEntryId(format!("original-{index}")),
origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
derivation: None,
},
})
.collect();
WorkerSessionStore::new(&source_root)
.unwrap()
.create_segment(
session,
segment,
&[LogEntry::AnnotatedSegmentStart {
ts: 1,
session_id: session,
system_prompt: None,
config: agen::llm_client::RequestConfig::default(),
history,
forked_from: None,
compacted_from: None,
}],
)
.unwrap();
let reader = session_store::public_material::SessionPublicMaterialReader::open_worker(
&source_root,
segment,
Default::default(),
)
.unwrap();
let ranges: Vec<_> = reader
.descriptors()
.map(|d| JournalSourceRange {
source: JournalSource {
session_id: session.to_string(),
pointer: JournalSourcePointer {
segment_id: d.identity.segment_id,
entry_ref: d.identity.entry_ref.clone(),
},
order: d.canonical_order,
content_sha256: d.sha256.clone(),
total_bytes: d.total_bytes,
},
bytes: JournalByteRange {
start: 0,
end: d.total_bytes,
},
})
.collect();
assert_eq!(ranges.len(), 2);
let attr = store
.record_session_attribution(
crate::subjektiv::SubjectSessionAttribution::new(
&subject.id,
RUNTIME,
&source_worker.worker_id,
session.to_string(),
)
.unwrap(),
)
.unwrap();
let writer = store
.claim_journal_writer(&subject.id, RUNTIME, &source_worker.worker_id)
.unwrap();
let host = store.bind_journal_session(&writer, attr.clone()).unwrap();
let journal_id = store
.accept_journal_sources(&host, "original activity", &ranges)
.unwrap();
let binding = subjektiv::api::JobAttemptBinding {
job_id: "fixture-journal-generation".into(),
attempt_id: "fixture-attempt".into(),
};
let generation = store
.prepare_journal_generation(&host, &journal_id, binding.clone(), ranges.clone())
.unwrap();
let job = JournalJobContext::validated(&store, &host, &generation.id, binding).unwrap();
for (descriptor, range) in reader.descriptors().zip(&ranges) {
subjektiv::journal::service::execute_bounded(
&store,
&job,
JournalJobRequest::ReadText {
read: JournalTextRead {
target: JournalTextTarget::Source {
pointer: range.source.pointer.clone(),
},
offset: 0,
limit: MAX_JOURNAL_READ_BYTES,
content_token: None,
},
},
|source, start, limit| {
assert_eq!(source, &range.source);
let page = reader.read_content(descriptor, start, limit).unwrap();
assert_eq!(page.next_offset, range.bytes.end);
Ok(subjektiv::journal::recording::JournalMaterial {
range: range.clone(),
text: page.content,
})
},
)
.unwrap();
}
let receipt = store
.execute_journal(
&job,
JournalOperation {
request_id: "initial-open".into(),
expected_state: generation.initial_journal.state_token,
action: JournalAction::Create {
title: "Choosing the approach".into(),
body: JOURNAL_BODY.into(),
ongoing: "Validation pending".into(),
state: JournalEntryState::Open,
source_end: None,
correction_refs: vec![],
},
covered_ranges: vec![ranges[0].clone()],
},
)
.unwrap();
let entry_id = receipt.entry_id.unwrap();
let kind = memory::extract::CandidateKind::Lesson;
let evidence_id = ranges[0].source.pointer.entry_ref.clone();
let candidate = memory::extract::StagingRecord::from_candidate(
"extracted-candidate",
"extraction-run",
memory::schema::SourceRef {
segment_id: segment.to_string(),
range: [0, 0],
},
memory::extract::ExtractedCandidate {
kind: kind.clone(),
claim: "Check the constraint before choosing".into(),
why_useful: "Avoid repeating the rejected approach".into(),
staleness: None,
evidence_ids: vec![evidence_id.clone()],
},
vec![memory::extract::StagingEvidence {
id: evidence_id.clone(),
kind: memory::schema::EvidenceKind::new("message"),
entry_range: Some([0, 0]),
origin: Some(serde_json::from_value(json!({"kind":"human_input"})).unwrap()),
excerpt: Some("Choose B after checking the constraint".into()),
summary: None,
}],
vec![memory::schema::SourceEvidenceRef {
session_id: Some(attr.session_id),
segment_id: Some(segment.to_string()),
entry_range: Some([0, 0]),
evidence_id: Some(evidence_id),
origin: Some(serde_json::from_value(json!({"kind":"human_input"})).unwrap()),
evidence_kind: Some(memory::schema::EvidenceKind::new("message")),
..Default::default()
}],
);
let (staged, _) = store
.stage_candidate_with_attribution(
crate::subjektiv::SubjectStagingRecord::attach(&subject.id, candidate),
store
.session_attribution(&session.to_string())
.unwrap()
.unwrap(),
)
.unwrap();
let (memory, _) = store
.apply_candidate(
&subject.id,
&staged.id,
crate::subjektiv::MemoryChangeTarget::Create,
crate::subjektiv::MemoryDraft::active(
kind,
&staged.claim,
"Check before choosing.",
&staged.why_useful,
"Adopt the extracted lesson",
),
"Accepted after comparison",
)
.unwrap();
let selector = MemoryJournalSelector {
memory_id: memory.id,
change_id: memory.change_id,
candidate_id: staged.id,
};
Self {
_dir: dir,
api,
identity,
store,
host,
job,
worker,
source_worker,
source_root,
selector,
journal_id,
entry_id,
ranges,
}
}
fn body_worker(api: &WorkspaceApi, subject: &str) -> RuntimeWorkerRef {
let memory = api
.config_store
.get_workspace_memory_settings(TEST_WORKSPACE_ID)
.unwrap();
let key = crate::subjektiv::subject_worker_singleton_key(subject).unwrap();
let reservation = api
.config_store
.reserve_worker_create(
TEST_WORKSPACE_ID,
RUNTIME,
&format!("body-{subject}"),
&format!("body-input-{subject}"),
Some(&key),
&memory,
)
.unwrap();
let worker = RuntimeWorkerRef::new(RUNTIME, reservation.worker_id.to_string());
seed_worker_source_member(api, RUNTIME, &worker.worker_id);
api.config_store
.complete_worker_create_reservation(TEST_WORKSPACE_ID, reservation.worker_id)
.unwrap();
worker
}
fn request(&self, worker: Option<&str>, request: &MemoryJournalRequest) -> Request<Body> {
runtime_source_request(
&self.identity,
worker,
"POST",
&route_path(),
serde_json::to_vec(request).unwrap(),
)
}
async fn call_as(
&self,
worker: &RuntimeWorkerRef,
request: MemoryJournalRequest,
) -> (StatusCode, Vec<u8>) {
response(
build_router(self.api.clone()),
self.request(Some(&worker.worker_id), &request),
)
.await
}
async fn call(&self, request: MemoryJournalRequest) -> MemoryJournalResponse {
let (status, bytes) = self.call_as(&self.worker, request).await;
assert_eq!(
status,
StatusCode::OK,
"{}",
String::from_utf8_lossy(&bytes)
);
serde_json::from_slice(&bytes).unwrap()
}
fn fix_initial(&self) {
self.store
.execute_journal(
&self.job,
JournalOperation {
request_id: "fix-initial".into(),
expected_state: self
.store
.read_journal(&self.host, &self.journal_id)
.unwrap()
.state_token,
action: JournalAction::Fix {
entry_id: self.entry_id.clone(),
source_end: self.ranges[0].source.pointer.clone(),
},
covered_ranges: vec![self.ranges[0].clone()],
},
)
.unwrap();
}
fn add_correction(&self) -> String {
self.store
.execute_journal(
&self.job,
JournalOperation {
request_id: "later-correction".into(),
expected_state: self
.store
.read_journal(&self.host, &self.journal_id)
.unwrap()
.state_token,
action: JournalAction::Create {
title: "Correction".into(),
body: "The chosen approach still needs validation.".into(),
ongoing: String::new(),
state: JournalEntryState::Fixed,
source_end: Some(self.ranges[1].source.pointer.clone()),
correction_refs: vec![self.entry_id.clone()],
},
covered_ranges: vec![self.ranges[1].clone()],
},
)
.unwrap()
.entry_id
.unwrap()
}
fn read_request(&self) -> MemoryJournalRequest {
MemoryJournalRequest::Read {
selector: self.selector.clone(),
source_index: 0,
journal_id: self.journal_id.clone(),
entry_id: self.entry_id.clone(),
expected_token: None,
byte_offset: 0,
max_bytes: 4096,
}
}
}
#[tokio::test]
async fn memory_journal_router_requires_signed_bound_worker_not_user_or_forged_identity() {
let f = Fixture::new().await;
let request = MemoryJournalRequest::Links {
selector: f.selector.clone(),
};
let body = serde_json::to_vec(&request).unwrap();
let app = build_router(f.api.clone());
let anonymous = Request::builder()
.method("POST")
.uri(route_path())
.header(CONTENT_TYPE, "application/json")
.body(Body::from(body.clone()))
.unwrap();
assert_eq!(
response(app.clone(), anonymous).await.0,
StatusCode::UNAUTHORIZED
);
let forged = Request::builder()
.method("POST")
.uri(route_path())
.header(CONTENT_TYPE, "application/json")
.header(
worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
"forged",
)
.body(Body::from(body.clone()))
.unwrap();
assert_eq!(
response(app.clone(), forged).await.0,
StatusCode::UNAUTHORIZED
);
let token = seed_test_workspace_owner_token(f.api.store.as_ref(), "memory-journal-user").await;
let user = Request::builder()
.method("POST")
.uri(route_path())
.header(CONTENT_TYPE, "application/json")
.header("authorization", format!("Bearer {token}"))
.body(Body::from(body))
.unwrap();
assert_eq!(response(app.clone(), user).await.0, StatusCode::FORBIDDEN);
assert_eq!(
response(app, f.request(None, &request)).await.0,
StatusCode::FORBIDDEN
);
}
#[tokio::test]
async fn memory_journal_router_default_worker_and_job_have_no_subject_capability() {
let f = Fixture::new().await;
let request = MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
};
let ordinary = RuntimeWorkerRef::new(RUNTIME, Uuid::new_v4().to_string());
seed_worker_source_member(&f.api, RUNTIME, &ordinary.worker_id);
let (status, bytes) = f.call_as(&ordinary, request.clone()).await;
assert_eq!(
status,
StatusCode::FORBIDDEN,
"{}",
String::from_utf8_lossy(&bytes)
);
// Even a singleton owner cannot reuse its body authority through a restricted Job binding.
let job = BackendJobRequest {
job_id: "restricted-job".into(),
purpose: "test".into(),
input_ref: "fixture://job".into(),
input: json!({}),
instruction: "No Memory Journal grant".into(),
profile: "builtin:backend-job".into(),
grants: Default::default(),
serialization_key: None,
source_worker: None,
notification_target: None,
limits: Default::default(),
};
let now = Utc::now().to_rfc3339();
let reserved = f
.api
.store
.reserve_backend_job(TEST_WORKSPACE_ID, &job, &now)
.unwrap();
f.api
.store
.claim_backend_job_attempt_dispatch(
TEST_WORKSPACE_ID,
&job.job_id,
&reserved.attempt.attempt_id,
&now,
)
.unwrap();
f.api
.store
.bind_backend_job_attempt_worker(
TEST_WORKSPACE_ID,
&job.job_id,
&reserved.attempt.attempt_id,
&f.worker,
None,
&now,
)
.unwrap();
let (status, bytes) = f.call_as(&f.worker, request).await;
assert_eq!(
status,
StatusCode::FORBIDDEN,
"{}",
String::from_utf8_lossy(&bytes)
);
let reader = JournalHostContext::validated_reader(
&f.store,
&f.store
.read_journal(&f.host, &f.journal_id)
.unwrap()
.subject_id,
)
.unwrap();
let MemoryJournalResponse::Links(links) = f
.store
.execute_memory_journal(
&reader,
MemoryJournalRequest::Links {
selector: f.selector.clone(),
},
)
.unwrap()
else {
panic!("links required")
};
assert_eq!(
links.sources[0].connection,
SourceConnection::Unconnected {
reason: UnconnectedReason::NotReconciled
}
);
}
#[tokio::test]
async fn memory_journal_router_reconciles_only_fixed_events_and_retries_without_duplication() {
let f = Fixture::new().await;
let reconcile = MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
};
let MemoryJournalResponse::Links(open) = f.call(reconcile.clone()).await else {
panic!("links required")
};
assert_eq!(
open.sources[0].connection,
SourceConnection::Unconnected {
reason: UnconnectedReason::JournalNotFixed
}
);
assert_eq!(
f.store
.list_journal_entries(&f.host, &f.journal_id, None, 32)
.unwrap()
.items[0]
.state,
JournalEntryState::Open
);
f.fix_initial();
// Reading stored state alone must not silently trigger reconciliation.
assert_eq!(
f.call(MemoryJournalRequest::Links {
selector: f.selector.clone()
})
.await,
MemoryJournalResponse::Links(open)
);
let linked = f.call(reconcile.clone()).await;
assert_eq!(f.call(reconcile).await, linked);
assert_eq!(
f.call(MemoryJournalRequest::Links {
selector: f.selector.clone()
})
.await,
linked
);
let MemoryJournalResponse::Links(links) = linked else {
panic!("links required")
};
assert_eq!(links.sources.len(), 1);
assert_eq!(
links.sources[0].connection,
SourceConnection::Linked {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
target: JournalTargetAvailability::Available,
reconsideration: false,
}
);
assert_eq!(
f.store
.list_journal_entries(&f.host, &f.journal_id, None, 32)
.unwrap()
.items
.len(),
1
);
}
#[tokio::test]
async fn memory_journal_router_reads_saved_text_after_original_session_and_source_worker_removal() {
let f = Fixture::new().await;
f.fix_initial();
f.call(MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
})
.await;
assert!(f.source_root.exists());
fs::remove_dir_all(&f.source_root).unwrap();
f.api
.store
.delete_worker_registry(TEST_WORKSPACE_ID, &f.source_worker)
.unwrap();
assert!(!f.source_root.exists());
assert!(
f.api
.store
.get_worker_registry(TEST_WORKSPACE_ID, &f.source_worker)
.unwrap()
.is_none()
);
// No Runtime adapter exists for RUNTIME, so a restore/history dependency would fail.
let MemoryJournalResponse::Read(read) = f.call(f.read_request()).await else {
panic!("read required")
};
assert_eq!(
read.source.original_source,
OriginalSourceAvailability::Missing
);
assert_eq!(read.page.unwrap().text, JOURNAL_BODY);
let MemoryJournalResponse::Related(related) = f
.call(MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
after: None,
limit: 32,
})
.await
else {
panic!("related required")
};
assert_eq!(related.items.len(), 1);
assert_eq!(related.items[0].selector, f.selector);
assert!(!related.items[0].reconsideration);
assert!(related.next.is_none());
}
#[tokio::test]
async fn memory_journal_router_ids_do_not_grant_cross_subject_reads_or_reconciliation() {
let f = Fixture::new().await;
f.fix_initial();
f.call(MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
})
.await;
let other = f
.store
.create_subject(crate::subjektiv::SubjectRole::new("other").unwrap())
.unwrap();
let other_worker = Fixture::body_worker(&f.api, &other.id);
for request in [
MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
},
MemoryJournalRequest::Links {
selector: f.selector.clone(),
},
f.read_request(),
MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: f.entry_id.clone(),
after: None,
limit: 32,
},
] {
let (status, bytes) = f.call_as(&other_worker, request).await;
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"{}",
String::from_utf8_lossy(&bytes)
);
assert!(!String::from_utf8_lossy(&bytes).contains(JOURNAL_BODY));
}
// A model cannot override the authenticated Subject with an additional JSON field.
let mut body = serde_json::to_value(f.read_request()).unwrap();
body["subject_id"] = json!(other.id);
let request = runtime_source_request(
&f.identity,
Some(&f.worker.worker_id),
"POST",
&route_path(),
serde_json::to_vec(&body).unwrap(),
);
assert_eq!(
response(build_router(f.api.clone()), request).await.0,
StatusCode::UNPROCESSABLE_ENTITY
);
}
#[tokio::test]
async fn memory_journal_router_correction_reverse_lookup_preserves_memory_and_original_link() {
let f = Fixture::new().await;
f.fix_initial();
f.call(MemoryJournalRequest::Reconcile {
selector: f.selector.clone(),
})
.await;
let subject = f
.store
.read_journal(&f.host, &f.journal_id)
.unwrap()
.subject_id;
let before =
serde_json::to_value(f.store.memory(&subject, &f.selector.memory_id).unwrap()).unwrap();
let correction = f.add_correction();
let MemoryJournalResponse::Related(related) = f
.call(MemoryJournalRequest::Related {
journal_id: f.journal_id.clone(),
entry_id: correction,
after: None,
limit: 32,
})
.await
else {
panic!("related required")
};
assert_eq!(related.items.len(), 1);
assert_eq!(related.items[0].selector, f.selector);
assert!(related.items[0].reconsideration);
let MemoryJournalResponse::Read(read) = f.call(f.read_request()).await else {
panic!("read required")
};
assert_eq!(read.page.unwrap().text, JOURNAL_BODY);
assert!(matches!(read.source.connection, SourceConnection::Linked {
entry_id, reconsideration:true, ..
} if entry_id == f.entry_id));
assert_eq!(
serde_json::to_value(f.store.memory(&subject, &f.selector.memory_id).unwrap()).unwrap(),
before
);
}
@@ -1,5 +1,5 @@
// Production generated routes, SQLite and WebAuthn verifier share the passkey
// security fixture. Only the authenticator is SoftPasskey, not a browser/OS key.
// security fixture. The authenticator is a resident model, not a browser/OS key.
// The ordinary Server test target selects these isolated integration cases.
impl Fixture {
@@ -0,0 +1,335 @@
// The generated HTTP router, real SQLite and production verifier/session issuer
// are connected. Only the authenticator is the shared resident software model;
// SoftPasskey alone is not resident-capable and browser-native UX is not claimed.
impl Fixture {
async fn discoverable_options(&self) -> PasskeyLoginOptionsResponse {
let (status, _, body) = self
.post(
"/api/auth/passkeys/login/options",
None,
Some(ORIGIN_URL),
serde_json::json!({}),
)
.await;
assert_eq!(status, StatusCode::OK, "{body}");
let options: PasskeyLoginOptionsResponse = serde_json::from_value(body).unwrap();
assert!(options.public_key.public_key.allow_credentials.is_empty());
assert!(options.public_key.mediation.is_none());
assert_eq!(
serde_json::to_value(&options.public_key).unwrap()["publicKey"]["userVerification"],
"required"
);
let owner: Option<String> = self
.api
.config_store
.with_conn(|conn| {
Ok(conn.query_row(
"SELECT user_id FROM auth_challenges WHERE challenge_id=?1",
[&options.challenge_id],
|row| row.get(0),
)?)
})
.unwrap();
assert!(
owner.is_none(),
"normal challenge must not preselect any account"
);
options
}
async fn discoverable_request(&self, key: &mut Authenticator) -> PasskeyLoginCompleteRequest {
let options = self.discoverable_options().await;
let credential = key
.do_authentication(Url::parse(ORIGIN_URL).unwrap(), options.public_key)
.unwrap();
PasskeyLoginCompleteRequest {
challenge_id: options.challenge_id,
credential,
}
}
async fn discoverable_login(&self, key: &mut Authenticator) -> AuthenticatedUser {
let request = self.discoverable_request(key).await;
let (status, headers, body) = self
.post(
"/api/auth/passkeys/login/complete",
None,
Some(ORIGIN_URL),
serde_json::to_value(request).unwrap(),
)
.await;
assert_eq!(status, StatusCode::OK, "{body}");
let user: AuthUserResponse = serde_json::from_value(body).unwrap();
let cookie = headers[axum::http::header::SET_COOKIE]
.to_str()
.unwrap()
.split(';')
.next()
.unwrap();
let token = cookie.split_once('=').unwrap().1;
let live = crate::auth::resolve_request_actor_sync(
self.api.store.as_ref(),
None,
Some(token.to_string()),
)
.unwrap()
.unwrap();
assert_eq!(live.user_id, user.user.user_id);
assert_eq!(live.account_id, user.user.account_id);
user.user
}
}
#[tokio::test]
async fn resident_selection_with_multiple_accounts_and_all_enrollment_purposes_issues_only_owner_session()
{
let f = Fixture::new().await;
let mut device = authenticator();
let (alice_cookie, alice) = f.enroll("registration", None, "alice", &mut device).await;
let alice_id = f.credentials(&alice)[0].credential_id.clone();
let (_, bob) = f.enroll("registration", None, "bob", &mut device).await;
let bob_id = f.credentials(&bob)[0].credential_id.clone();
for (id, expected) in [(alice_id, alice.clone()), (bob_id, bob)] {
let id: webauthn_rs::prelude::Base64UrlSafeData =
serde_json::from_value(Value::String(id)).unwrap();
device.select(&id);
assert_eq!(f.discoverable_login(&mut device).await, expected);
}
let mut added = authenticator();
f.enroll("addition", Some(&alice_cookie), "", &mut added)
.await;
assert_eq!(f.discoverable_login(&mut added).await, alice);
// Registration policy controls account creation, not login or own addition.
let admin = f.activation_admin().await;
assert_eq!(f.discoverable_login(&mut added).await, alice);
let issued = f.issue_activation(&admin, "pending-discoverable").await;
let mut activated = authenticator();
let options = f.activation_options(&issued).await;
let body = Fixture::credential(options, &mut activated);
let (status, _, body) = f
.post(
"/api/auth/activation/complete",
None,
Some(ORIGIN_URL),
body,
)
.await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(f.discoverable_login(&mut activated).await, issued.user);
}
#[tokio::test]
async fn discoverable_verifier_rejects_unknown_owner_signature_origin_rp_uv_expiry_reuse_and_revocation()
{
let f = Fixture::new().await;
let mut key = authenticator();
let (_, alice) = f.enroll("registration", None, "alice", &mut key).await;
for defect in [
"unknown",
"handle",
"signature",
"origin",
"rp",
"uv",
"expired",
"consumed",
"stored_id",
"stored_owner",
] {
let mut request = f.discoverable_request(&mut key).await;
let original = f.credentials(&alice)[0].clone();
let change_byte =
|value: &mut webauthn_rs::prelude::Base64UrlSafeData, index: usize, bit: u8| {
let mut bytes = value.to_vec();
bytes[index] ^= bit;
*value = bytes.into();
};
match defect {
"unknown" => change_byte(&mut request.credential.raw_id, 0, 1),
"handle" => request.credential.response.user_handle = Some(vec![0; 16].into()),
"signature" => change_byte(&mut request.credential.response.signature, 0, 1),
"origin" => {
let mut data: Value =
serde_json::from_slice(&request.credential.response.client_data_json).unwrap();
data["origin"] = Value::String("https://other.test".into());
request.credential.response.client_data_json =
serde_json::to_vec(&data).unwrap().into();
}
"rp" => change_byte(&mut request.credential.response.authenticator_data, 0, 1),
"uv" => change_byte(&mut request.credential.response.authenticator_data, 32, 4),
"expired" => f.expire_challenge(&request.challenge_id),
"consumed" => {
f.api
.store
.consume_auth_challenge_by_id(
&request.challenge_id,
"passkey_login",
&crate::auth::now_rfc3339(),
)
.unwrap();
}
"stored_id" => {
let mut corrupt = original.clone();
let mut value: Value = serde_json::from_str(&corrupt.public_key_cose).unwrap();
// Keep the indexed ID, replace the embedded verified credential ID.
value["cred"]["cred_id"] = Value::String("YW5vdGhlci1jcmVkZW50aWFs".into());
corrupt.public_key_cose = serde_json::to_string(&value).unwrap();
f.api.store.upsert_passkey_credential(&corrupt).unwrap();
}
"stored_owner" => {
let (_, bob) = f
.enroll("registration", None, "bob", &mut authenticator())
.await;
f.api
.config_store
.with_conn(|conn| {
conn.execute(
"UPDATE passkey_credentials SET user_id=?1 WHERE credential_id=?2",
[&bob.user_id, &original.credential_id],
)?;
Ok(())
})
.unwrap();
}
_ => unreachable!(),
}
let (status, headers, body) = f
.post(
"/api/auth/passkeys/login/complete",
None,
Some(ORIGIN_URL),
serde_json::to_value(&request).unwrap(),
)
.await;
assert!(status.is_client_error(), "{defect}: {body}");
assert!(
!headers.contains_key(axum::http::header::SET_COOKIE),
"{defect}"
);
let (status, headers, _) = f
.post(
"/api/auth/passkeys/login/complete",
None,
Some(ORIGIN_URL),
serde_json::to_value(request).unwrap(),
)
.await;
assert!(
status.is_client_error(),
"failed attempts must consume their ceremony: {defect}"
);
assert!(!headers.contains_key(axum::http::header::SET_COOKIE));
f.api
.config_store
.with_conn(|conn| {
conn.execute(
"UPDATE passkey_credentials SET user_id=?1 WHERE credential_id=?2",
[&original.user_id, &original.credential_id],
)?;
Ok(())
})
.unwrap();
f.api.store.upsert_passkey_credential(&original).unwrap();
}
assert_eq!(f.discoverable_login(&mut key).await, alice);
}
#[tokio::test]
async fn signed_assertions_with_foreign_origin_rp_or_without_uv_are_rejected_by_required_verifier()
{
let f = Fixture::new().await;
let mut key = authenticator();
let (_, alice) = f.enroll("registration", None, "alice", &mut key).await;
let id: webauthn_rs::prelude::Base64UrlSafeData = serde_json::from_value(Value::String(
f.credentials(&alice)[0].credential_id.clone(),
))
.unwrap();
for defect in ["origin", "rp", "uv"] {
let mut options = f.discoverable_options().await;
let mut origin = ORIGIN_URL;
match defect {
"origin" => origin = "https://sub.workspace.test",
"rp" => {
origin = "https://other.test";
options.public_key.public_key.rp_id = "other.test".into();
// Explicit signer selection produces a genuinely signed assertion
// for the wrong RP; normal Server options remain empty/unchanged.
options.public_key.public_key.allow_credentials =
vec![webauthn_rs_proto::AllowCredentials {
type_: "public-key".into(),
id: id.clone(),
transports: None,
}];
}
"uv" => {
options.public_key.public_key.user_verification =
webauthn_rs_proto::UserVerificationPolicy::Preferred
}
_ => unreachable!(),
}
let mut credential = key
.do_authentication(Url::parse(origin).unwrap(), options.public_key)
.unwrap();
if defect == "uv" {
assert_eq!(credential.response.authenticator_data[32] & 4, 0);
}
credential.response.user_handle = Some(
passkey_user_handle(&alice.account_id)
.as_bytes()
.to_vec()
.into(),
);
let request = PasskeyLoginCompleteRequest {
challenge_id: options.challenge_id,
credential,
};
rejected(
f.post(
"/api/auth/passkeys/login/complete",
None,
Some(ORIGIN_URL),
serde_json::to_value(request).unwrap(),
)
.await,
StatusCode::BAD_REQUEST,
"webauthn_login_verification_failed",
);
}
assert_eq!(f.discoverable_login(&mut key).await, alice);
}
#[tokio::test]
async fn non_resident_library_authenticator_refuses_registration_without_creating_identity() {
use webauthn_authenticator_rs::{WebauthnAuthenticator, softpasskey::SoftPasskey};
let f = Fixture::new().await;
let options = f.options("registration", None, "unsupported-key").await;
let selection = options
.public_key
.public_key
.authenticator_selection
.as_ref()
.unwrap();
assert!(selection.require_resident_key);
assert_eq!(
selection.resident_key,
Some(webauthn_rs_proto::ResidentKeyRequirement::Required)
);
assert_eq!(
selection.user_verification,
webauthn_rs_proto::UserVerificationPolicy::Required
);
let result = WebauthnAuthenticator::new(SoftPasskey::new(true))
.do_registration(Url::parse(ORIGIN_URL).unwrap(), options.public_key);
assert!(
result.is_err(),
"SoftPasskey itself must not be reported as resident-capable"
);
assert!(
f.api
.store
.get_user_by_handle("unsupported-key")
.unwrap()
.is_none()
);
}
+368 -4
View File
@@ -119,7 +119,7 @@ pub use ticket::{
};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 100;
const LATEST_SCHEMA_VERSION: i64 = 101;
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";
@@ -427,6 +427,11 @@ const MIGRATIONS: &[Migration] = &[
name: "Provisional users and scoped passkey activation",
apply: activation::migrate,
},
Migration {
version: 101,
name: "Durable explicit Subject body release",
apply: migrate_subject_release_v100_to_v101,
},
];
#[derive(Clone, Copy)]
@@ -1954,6 +1959,15 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
workspace_id: &str,
singleton_key: &str,
) -> Result<Option<WorkerSingletonLease>>;
fn retained_worker_singleton_lease(
&self,
_workspace_id: &str,
_worker: &RuntimeWorkerRef,
) -> Result<Option<WorkerSingletonLease>> {
Err(Error::Store(
"retained singleton lookup is unavailable".into(),
))
}
fn require_current_worker_singleton_owner(
&self,
workspace_id: &str,
@@ -3164,6 +3178,156 @@ impl SqliteWorkspaceStore {
})
}
/// Retained identity for lifecycle/retention reconciliation only, never execution authority.
pub(crate) fn retained_worker_singleton_lease(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
) -> Result<Option<WorkerSingletonLease>> {
self.with_conn(|conn| {
let row: Option<(Option<String>, Option<i64>)> = conn.query_row(
"SELECT singleton_key,singleton_generation FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",
params![workspace_id,worker.runtime_id,worker.worker_id], |r| Ok((r.get(0)?,r.get(1)?)),
).optional()?;
match row {
Some((Some(key),Some(generation))) => Ok(Some(WorkerSingletonLease {
key, generation: generation.try_into().map_err(|_| Error::Store("invalid singleton generation".into()))?, worker:worker.clone(),
})),
Some((None,None)) | None => Ok(None),
_ => Err(Error::Store("incomplete singleton lease".into())),
}
})
}
pub(crate) fn retained_subject_release_operation(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
) -> Result<Option<String>> {
self.with_conn(|c| Ok(c.query_row("SELECT operation_id FROM subject_worker_releases WHERE workspace_id=?1 AND worker_id=?2 AND runtime_id=?3 AND singleton_generation=?4 AND binding_id=?5 AND cancelled=0",params![workspace,lease.worker.worker_id,lease.worker.runtime_id,lease.generation as i64,binding],|r|r.get(0)).optional()?))
}
pub(crate) fn subject_release_pending(
&self,
workspace: &str,
worker: &RuntimeWorkerRef,
) -> Result<bool> {
self.with_conn(|c| Ok(c.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND cancelled=0 AND completed=0)",params![workspace,worker.runtime_id,worker.worker_id],|r|r.get(0))?))
}
pub(crate) fn require_subject_release_intent(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
) -> Result<()> {
self.with_conn(|c| {
let active: bool = c.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND binding_id=?6 AND operation_id=?7 AND cancelled=0)",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.key,lease.generation as i64,binding,operation],|r|r.get(0))?;
if !active {return Err(Error::RepositoryConflict("Subject release intent absent, cancelled or mismatched".into()));} Ok(())
})
}
pub(crate) fn reserve_subject_release(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
) -> Result<()> {
if binding.trim().is_empty()
|| operation.trim().is_empty()
|| binding.len() > 512
|| operation.len() > 512
{
return Err(Error::InvalidInput(
"invalid Subject release identity".into(),
));
}
self.with_conn_mut(|c| {
let tx=c.transaction_with_behavior(TransactionBehavior::Immediate)?;
let saved: Option<(String,String,String,u64,String,bool)> = tx.query_row("SELECT runtime_id,worker_id,singleton_key,singleton_generation,binding_id,cancelled FROM subject_worker_releases WHERE workspace_id=?1 AND operation_id=?2",params![workspace,operation],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?,r.get(4)?,r.get(5)?))).optional()?;
if let Some(saved)=saved {
if saved != (lease.worker.runtime_id.clone(),lease.worker.worker_id.clone(),lease.key.clone(),lease.generation,binding.into(),false) {return Err(Error::RepositoryConflict("release operation mismatched or terminally cancelled; use a new operation".into()));} return Ok(());
}
require_current_worker_singleton_lease(&tx,workspace,lease)?;
let busy: bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND worker_id=?2 AND cancelled=0)",params![workspace,lease.worker.worker_id],|r|r.get(0))?;
if busy {return Err(Error::RepositoryConflict("another Subject release intent is active".into()));}
tx.execute("INSERT INTO subject_worker_releases(workspace_id,runtime_id,worker_id,singleton_key,singleton_generation,binding_id,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.key,lease.generation as i64,binding,operation])?;
tx.commit()?; Ok(())
})
}
pub(crate) fn cancel_subject_release(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
) -> Result<()> {
self.with_conn(|c| {
let changed=c.execute("UPDATE subject_worker_releases SET cancelled=1 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_generation=?4 AND binding_id=?5 AND operation_id=?6 AND proof IS NULL AND completed=0",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.generation as i64,binding,operation])?;
if changed!=1 {return Err(Error::RepositoryConflict("cannot confirm cancellation of different or committed Subject release".into()));} Ok(())
})
}
pub(crate) fn subject_release_proven(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
) -> Result<bool> {
self.with_conn(|c|Ok(c.query_row("SELECT EXISTS(SELECT 1 FROM subject_worker_releases WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND binding_id=?6 AND operation_id=?7 AND proof IS NOT NULL AND cancelled=0)",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.key,lease.generation as i64,binding,operation],|r|r.get(0))?))
}
pub(crate) fn confirm_subject_release_proof(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
erased: bool,
) -> Result<()> {
let proof = if erased {
"erased"
} else {
"runtime_disconnected"
};
self.with_conn(|c| {
let changed=c.execute("UPDATE subject_worker_releases SET proof=?8 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND binding_id=?6 AND operation_id=?7 AND cancelled=0 AND (proof IS NULL OR proof=?8)",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.key,lease.generation as i64,binding,operation,proof])?;
if changed!=1 {return Err(Error::RepositoryConflict("Subject release proof does not match active intent".into()));} Ok(())
})
}
pub(crate) fn release_subject_singleton(
&self,
workspace: &str,
lease: &WorkerSingletonLease,
binding: &str,
operation: &str,
) -> Result<()> {
self.require_subject_release_intent(workspace, lease, binding, operation)?;
if !lease
.key
.starts_with(crate::subjektiv::SUBJECT_WORKER_SINGLETON_PREFIX)
{
return Err(Error::InvalidInput("not a Subject singleton".into()));
}
self.with_conn_mut(|c| {
let tx=c.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (proof,completed,cancelled):(Option<String>,bool,bool)=tx.query_row("SELECT proof,completed,cancelled FROM subject_worker_releases WHERE workspace_id=?1 AND operation_id=?2",params![workspace,operation],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?;
if cancelled || proof.is_none() {return Err(Error::RepositoryConflict("Subject release lacks confirmed proof; retain occupancy".into()));}
if completed {return Ok(());}
require_current_worker_singleton_lease(&tx,workspace,lease)?;
tx.execute("UPDATE subject_worker_releases SET completed=1 WHERE workspace_id=?1 AND operation_id=?2 AND cancelled=0",params![workspace,operation])?;
let changed=tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?6 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND singleton_key=?4 AND singleton_generation=?5 AND state IN ('reserved','created','removed')",params![workspace,lease.worker.runtime_id,lease.worker.worker_id,lease.key,lease.generation as i64,chrono::Utc::now().to_rfc3339()])?;
if changed!=1 {return Err(Error::RepositoryConflict("Subject owner changed before release".into()));}
tx.commit()?; Ok(())
})
}
pub(crate) fn current_worker_singleton_owner(
&self,
workspace_id: &str,
@@ -7931,6 +8095,14 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
SqliteWorkspaceStore::current_worker_singleton_owner(self, workspace_id, singleton_key)
}
fn retained_worker_singleton_lease(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
) -> Result<Option<WorkerSingletonLease>> {
SqliteWorkspaceStore::retained_worker_singleton_lease(self, workspace_id, worker)
}
fn require_current_worker_singleton_owner(
&self,
workspace_id: &str,
@@ -14840,6 +15012,22 @@ fn migrate_result_reports_v97_to_v98(conn: &Connection) -> Result<()> {
Ok(())
}
fn migrate_subject_release_v100_to_v101(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 100 {
return Err(Error::Store(
"Subject release migration requires schema 100".into(),
));
}
let tx = conn.unchecked_transaction()?;
tx.execute_batch(include_str!("subject_release_schema.sql"))?;
tx.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(101,?1)",
["Durable explicit Subject body release"],
)?;
tx.commit()?;
Ok(())
}
fn migrate_hosting_policy_v98_to_v99(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 98 {
return Err(Error::Store(
@@ -15521,7 +15709,10 @@ fn current_worker_singleton_owner(
) -> Result<Option<(WorkerSingletonLease, Option<String>)>> {
let owner = conn
.query_row(
"SELECT owner.runtime_id, owner.worker_id, owner.generation, reservation.state \
"SELECT owner.runtime_id, owner.worker_id, owner.generation, \
CASE WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \
AND NOT EXISTS(SELECT 1 FROM subject_worker_releases r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.completed=1) \
THEN 'created' ELSE reservation.state END \
FROM worker_singleton_owners owner \
LEFT JOIN worker_create_reservations reservation \
ON reservation.workspace_id = owner.workspace_id \
@@ -18547,6 +18738,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
WorkspaceSchemaMigrationStep {version:98, name:"Assignment-independent implementation reports".into()},
WorkspaceSchemaMigrationStep {version:99, name:"Explicit hosting administrators and self-registration policy".into()},
WorkspaceSchemaMigrationStep {version:100, name:"Provisional users and scoped passkey activation".into()},
WorkspaceSchemaMigrationStep {version:101, name:"Durable explicit Subject body release".into()},
]
);
@@ -18657,6 +18849,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(98, "Assignment-independent implementation reports".to_string()),
(99, "Explicit hosting administrators and self-registration policy".to_string()),
(100, "Provisional users and scoped passkey activation".to_string()),
(101, "Durable explicit Subject body release".to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -20734,7 +20927,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.await
.unwrap();
}
let key = "subjektiv:主体 id/?!";
// Generic singleton lifecycle; Subject-prefixed keys have a separate explicit release contract.
let key = "fixture:主体 id/?!";
let memory_a = store.get_workspace_memory_settings("workspace-a").unwrap();
let first = store
.reserve_worker_create(
@@ -20842,6 +21036,175 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
);
}
#[test]
fn subject_release_migration_rolls_back_schema_and_marker_then_reopens() {
let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db");
prepare_retained_schema(&database, 100);
let conn = Connection::open(&database).unwrap();
configure_sqlite(&conn).unwrap();
conn.execute_batch("CREATE TRIGGER interrupt_subject_release_migration BEFORE INSERT ON __yoi_schema_migrations WHEN NEW.version=101 BEGIN SELECT RAISE(ABORT,'injected Subject release migration interruption'); END;").unwrap();
assert!(migrate_subject_release_v100_to_v101(&conn).is_err());
assert_eq!(current_schema_version(&conn).unwrap(), 100);
assert!(!table_exists(&conn, "subject_worker_releases").unwrap());
conn.execute_batch("DROP TRIGGER interrupt_subject_release_migration")
.unwrap();
migrate_subject_release_v100_to_v101(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 101);
assert!(table_exists(&conn, "subject_worker_releases").unwrap());
assert!(migrate_subject_release_v100_to_v101(&conn).is_err());
drop(conn);
let reopened = SqliteWorkspaceStore::open_for_test(database).unwrap();
reopened
.with_conn(|conn| verify_schema_history(conn, 101))
.unwrap();
}
#[tokio::test]
async fn subject_generic_removal_retains_occupancy_until_exact_explicit_release_receipt() {
let dir = tempfile::tempdir().unwrap();
let database = dir.path().join("server.db");
let store = SqliteWorkspaceStore::open_api_fixture(&database).unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".into(),
owner_account_id: "owner-account".into(),
display_name: "Workspace A".into(),
state: "active".into(),
created_at: "2026-10-01T00:00:00Z".into(),
updated_at: "2026-10-01T00:00:00Z".into(),
})
.await
.unwrap();
let memory = store.get_workspace_memory_settings("workspace-a").unwrap();
let key = "subjektiv:subject-a";
let first = store
.reserve_worker_create(
"workspace-a",
"runtime-a",
"first",
"sha256:first",
Some(key),
&memory,
)
.unwrap();
let lease = first.singleton.unwrap();
store
.fail_worker_create_reservation(
"workspace-a",
"runtime-a",
first.worker_id,
&first.create_fingerprint,
)
.unwrap();
assert_eq!(
store
.current_worker_singleton_owner("workspace-a", key)
.unwrap(),
Some(lease.clone())
);
assert!(
store
.reserve_worker_create(
"workspace-a",
"runtime-b",
"early",
"sha256:early",
Some(key),
&memory
)
.is_err()
);
store
.reserve_subject_release("workspace-a", &lease, "binding-a", "cancelled-first")
.unwrap();
assert!(
store
.subject_release_pending("workspace-a", &lease.worker)
.unwrap()
);
store
.cancel_subject_release("workspace-a", &lease, "binding-a", "cancelled-first")
.unwrap();
assert!(
!store
.subject_release_pending("workspace-a", &lease.worker)
.unwrap()
);
assert!(
store
.reserve_subject_release("workspace-a", &lease, "binding-a", "cancelled-first")
.is_err()
);
store
.reserve_subject_release("workspace-a", &lease, "binding-a", "release-a")
.unwrap();
drop(store);
let reopened = SqliteWorkspaceStore::open_for_test(&database).unwrap();
assert_eq!(
reopened
.current_worker_singleton_owner("workspace-a", key)
.unwrap(),
Some(lease.clone())
);
assert!(
reopened
.reserve_subject_release("workspace-a", &lease, "binding-later", "release-a")
.is_err()
);
// SQL component boundary: the Host's actual Controller and durable writer-revoke proof
// are tested at their integration boundaries, not certified by this private SQL helper.
assert!(
reopened
.release_subject_singleton("workspace-a", &lease, "binding-a", "release-a")
.is_err()
);
reopened
.confirm_subject_release_proof("workspace-a", &lease, "binding-a", "release-a", false)
.unwrap();
reopened
.release_subject_singleton("workspace-a", &lease, "binding-a", "release-a")
.unwrap();
assert!(
reopened
.current_worker_singleton_owner("workspace-a", key)
.unwrap()
.is_none()
);
let replacement = reopened
.reserve_worker_create(
"workspace-a",
"runtime-b",
"next",
"sha256:next",
Some(key),
&memory,
)
.unwrap();
assert_eq!(
replacement.singleton.unwrap().generation,
lease.generation + 1
);
reopened
.release_subject_singleton("workspace-a", &lease, "binding-a", "release-a")
.unwrap();
assert!(
reopened
.release_subject_singleton(
"workspace-a",
&lease,
"binding-a",
"different-operation"
)
.is_err()
);
assert!(
reopened
.require_current_worker_singleton_owner("workspace-a", &lease.worker)
.is_err()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn concurrent_worker_singleton_reservations_have_one_winner() {
let dir = tempfile::tempdir().unwrap();
@@ -21272,7 +21635,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(97, "MR notifications and authenticated implementation results".into()),
(98, "Assignment-independent implementation reports".into()),
(99, "Explicit hosting administrators and self-registration policy".into()),
(100, "Provisional users and scoped passkey activation".into())]);
(100, "Provisional users and scoped passkey activation".into()),
(101, "Durable explicit Subject body release".into())]);
for (table, rows) in unchanged.iter().zip(before.iter()) {
assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
}
@@ -0,0 +1,16 @@
CREATE TABLE subject_worker_releases (
workspace_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
singleton_key TEXT NOT NULL,
singleton_generation INTEGER NOT NULL CHECK (singleton_generation > 0),
binding_id TEXT NOT NULL,
operation_id TEXT NOT NULL,
proof TEXT CHECK (proof IN ('runtime_disconnected','erased')),
completed INTEGER NOT NULL DEFAULT 0 CHECK (completed IN (0,1)),
cancelled INTEGER NOT NULL DEFAULT 0 CHECK (cancelled IN (0,1)),
PRIMARY KEY (workspace_id, operation_id),
CHECK (cancelled = 0 OR (completed = 0 AND proof IS NULL)),
FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_create_reservations(workspace_id, worker_id) ON DELETE CASCADE
);
CREATE UNIQUE INDEX subject_worker_release_active ON subject_worker_releases(workspace_id,worker_id) WHERE cancelled=0;
@@ -94,6 +94,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
"workdir_registry",
"workdir_removal_operations",
"worker_control_grants",
"subject_worker_releases",
"worker_create_reservations",
"worker_diagnostics_archives",
"worker_mutation_source_proof_jtis",