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

feat: configure subject attachment on worker launch

Browse Source
This commit is contained in:
Hare
2026-10-04 08:40:30 +09:00
parent 1bbc48dad0
commit f7b722246d
25 changed files with 831 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/backend_runtime.rs
+3 -1
View File
@@ -819,7 +819,8 @@ mod tests {
"profiles": [{
"id": "builtin:default",
"label": "Default",
"description": ""
"description": "",
"feature_connections": { "subjektiv": false }
}],
"repositories": [],
"working_directories": [],
@@ -883,6 +884,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
};
crates/client/src/workspace_product.rs
+1
View File
@@ -532,6 +532,7 @@ impl BackendWorkspaceProductClient {
content: format!("Please handle intake for Ticket {ticket_id}."),
}],
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
};
let workspace_id = self.workspace_id.clone();
crates/server-api/src/lib.rs
+32 -22
View File
@@ -249,7 +249,6 @@ impl_openapi_schema!(
SubjektivMemoryQueryResponse,
SubjektivMemoryReadResponse,
SubjektivMemoryListRevisionsResponse,
SubjektivSubjectWorkerStartRequest,
SubjektivStageCandidateRequest,
SubjektivStageCandidateResponse,
SubjektivRecordSessionRequest,
@@ -1234,21 +1233,6 @@ pub trait ServerApi {
#[path] workspace_id: String,
#[path] subject_id: String,
) -> Result<SubjektivSubjectResponse, RepositoryApiError>;
#[post(
"/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/worker",
status = 200,
error_status = 400,
additional_error_statuses = [401, 403, 404, 409, 500, 502, 503],
bearer_auth = true,
browser_auth = true,
)]
async fn subjektiv_subject_worker_start(
&self,
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
#[path] subject_id: String,
#[body] request: SubjektivSubjectWorkerStartRequest,
) -> Result<BrowserCreateWorkerResponse, RepositoryApiError>;
#[post(
"/api/w/{workspace_id}/subjektiv/staging",
status = 200,
@@ -4780,12 +4764,6 @@ pub struct SubjektivMemoryRevisionsQuery {
pub cursor: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct SubjektivSubjectWorkerStartRequest {
pub worker: CreateWorkspaceWorkerRequest,
}
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct SubjektivStageCandidateRequest {
@@ -9489,6 +9467,14 @@ pub struct WorkerLaunchProfileCandidate {
pub id: String,
pub label: String,
pub description: String,
pub feature_connections: WorkerLaunchFeatureConnectionOptions,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkerLaunchFeatureConnectionOptions {
pub subjektiv: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
@@ -9520,6 +9506,22 @@ pub struct CreateWorkspaceWorkerTicketAssignmentRequest {
pub operation_id: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkspaceWorkerFeatureConnectionsRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
#[cfg_attr(feature = "typescript", ts(optional = nullable))]
pub subjektiv: Option<SubjektivWorkerConnectionRequest>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct SubjektivWorkerConnectionRequest {
pub subject_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
@@ -9537,6 +9539,9 @@ pub struct CreateWorkspaceWorkerRequest {
pub initial_submit: Vec<protocol::Segment>,
#[serde(default)]
pub workdir_attachments: Vec<BrowserWorkerWorkingDirectorySelection>,
/// Explicit product Feature connections. Backend validation turns these into
/// trusted host attachments; Runtime and Profile inputs cannot self-attest them.
pub feature_connections: WorkspaceWorkerFeatureConnectionsRequest,
/// Backend idempotency key used only for authenticated Worker-owned spawn/control.
#[serde(default)]
pub control_operation_id: Option<String>,
@@ -10703,11 +10708,14 @@ pub fn worker_launch_api_typescript() -> String {
RuntimeWorkerWorkdirAttachmentSummary::decl(&config),
WorkerLaunchWorkerSummary::decl(&config),
WorkerLaunchRuntimeOption::decl(&config),
WorkerLaunchFeatureConnectionOptions::decl(&config),
WorkerLaunchProfileCandidate::decl(&config),
WorkingDirectoryRepositoryOption::decl(&config),
WorkerLaunchOptionsResponse::decl(&config),
BrowserWorkerWorkingDirectorySelection::decl(&config),
CreateWorkspaceWorkerTicketAssignmentRequest::decl(&config),
SubjektivWorkerConnectionRequest::decl(&config),
WorkspaceWorkerFeatureConnectionsRequest::decl(&config),
CreateWorkspaceWorkerRequest::decl(&config),
BrowserCreateWorkerResponse::decl(&config),
BrowserWorkspaceOrchestratorResponse::decl(&config),
@@ -11748,6 +11756,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
})
.unwrap();
@@ -11761,6 +11770,7 @@ mod tests {
"ticket_assignment": null,
"initial_submit": [],
"workdir_attachments": [],
"feature_connections": {},
"control_operation_id": null,
})
);
crates/tui/src/backend_spawn.rs
+3
View File
@@ -167,6 +167,7 @@ fn request_from_selection(selection: Selection) -> BackendCreateWorkerRequest {
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
ticket_assignment: None,
feature_connections: Default::default(),
control_operation_id: None,
}
}
@@ -431,11 +432,13 @@ mod tests {
id: "builtin:default".to_string(),
label: "Default".to_string(),
description: String::new(),
feature_connections: Default::default(),
},
BackendWorkerLaunchProfileCandidate {
id: "builtin:coder".to_string(),
label: "Coder".to_string(),
description: "Ticket implementation".to_string(),
feature_connections: Default::default(),
},
],
default_profile: Some("builtin:coder".to_string()),
crates/worker/src/feature/builtin/manage_worker.rs
+1
View File
@@ -522,6 +522,7 @@ fn workspace_worker_create_request(
relative_cwd: attachment.relative_cwd,
})
.collect(),
feature_connections: Default::default(),
control_operation_id: Some(control_operation_id),
})
}
crates/workspace-server/src/profile_settings.rs
+17
View File
@@ -447,6 +447,23 @@ pub fn build_virtual_profile_config_bundle_with_prompt_projection(
Ok(Some(bundle))
}
pub(crate) fn resolve_profile_manifest_from_config_bundle(
bundle: &ConfigBundle,
selector: &str,
) -> Result<manifest::WorkerManifest> {
let archive = bundle.profile_source_archive.as_ref().ok_or_else(|| {
Error::Config(format!(
"resolved Profile {selector:?} is missing its source archive"
))
})?;
let verified = archive.verify().map_err(|error| {
Error::Config(format!("failed to verify Profile {selector:?}: {error}"))
})?;
verified
.resolve_profile(selector, Path::new("/"), "worker-launch-validation")
.map_err(|error| Error::Config(format!("failed to resolve Profile {selector:?}: {error}")))
}
fn build_virtual_profile_archive(
selector: &str,
entry: &VirtualProfileEntry,
crates/workspace-server/src/server.rs
+233 -122
View File
@@ -5468,9 +5468,6 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::subjektiv_subject_retire(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_subject_worker_start(
service.clone(),
))
.merge(server_api::server_api_axum::subjektiv_stage_candidate(
service.clone(),
))
@@ -7235,27 +7232,6 @@ impl server_api::ServerApi for ServerApiContractService {
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_subject_worker_start(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
subject_id: String,
request: server_api::SubjektivSubjectWorkerStartRequest,
) -> std::result::Result<BrowserCreateWorkerResponse, server_api::RepositoryApiError> {
scoped_start_subjektiv_subject_worker(
State(self.workspace_api()?.clone()),
AxumPath(ScopedSubjektivSubjectPath {
workspace_id,
subject_id,
}),
contract_request_headers(&context)?,
Json(request),
)
.await
.map(|Json(response)| response)
.map_err(ApiError::into_repository_api_error)
}
async fn subjektiv_stage_candidate(
&self,
context: server_api::ServerRequestContext,
@@ -16780,7 +16756,7 @@ fn resident_summary_output(
}
}
const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:";
const SUBJEKTIV_SINGLETON_PREFIX: &str = crate::subjektiv::SUBJECT_WORKER_SINGLETON_PREFIX;
const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str =
crate::hosts::SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX;
const SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE: &str = "subjektiv-memory-consolidation";
@@ -16821,15 +16797,6 @@ fn subjektiv_scope_from_singleton_key(key: &str) -> ApiResult<(&str, SubjektivWo
Ok((subject_id, authority))
}
fn subjektiv_singleton_key(subject_id: &str) -> ApiResult<String> {
if subject_id.trim().is_empty() || subject_id.chars().any(char::is_control) {
return Err(Error::InvalidInput("subjektiv subject id is invalid".to_string()).into());
}
let key = format!("{SUBJEKTIV_SINGLETON_PREFIX}{subject_id}");
crate::store::validate_worker_singleton_key(&key)?;
Ok(key)
}
fn is_dedicated_subjektiv_consolidation_worker(
worker: &InternalWorkerSummary,
workspace_id: &str,
@@ -16943,7 +16910,7 @@ fn subjektiv_subject_response(
api: &WorkspaceApi,
subject: crate::subjektiv::SubjectRecord,
) -> server_api::SubjektivSubjectResponse {
let current_worker = subjektiv_singleton_key(&subject.id)
let current_worker = crate::subjektiv::subject_worker_singleton_key(&subject.id)
.ok()
.and_then(|key| {
api.store
@@ -17199,36 +17166,6 @@ async fn scoped_retire_subjektiv_subject(
Ok(Json(subjektiv_subject_response(&api, subject)))
}
async fn scoped_start_subjektiv_subject_worker(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedSubjektivSubjectPath>,
headers: HeaderMap,
Json(request): Json<server_api::SubjektivSubjectWorkerStartRequest>,
) -> ApiResult<Json<BrowserCreateWorkerResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let subject = open_subjektiv_store(&api)?
.subject(&path.subject_id)
.map_err(|error| Error::Store(error.to_string()))?
.ok_or_else(|| Error::SubjektivSubjectNotFound(path.subject_id.clone()))?;
if subject.state != crate::subjektiv::SubjectState::Active {
return Err(Error::InvalidInput(format!(
"subjektiv subject `{}` is retired",
path.subject_id
))
.into());
}
let mut worker = request.worker;
if worker.singleton_key.is_some() {
return Err(Error::InvalidInput(
"subjektiv subject Worker start owns singleton selection; singleton_key must be omitted"
.to_string(),
)
.into());
}
worker.singleton_key = Some(subjektiv_singleton_key(&path.subject_id)?);
create_workspace_worker_inner(api, headers, worker, None, true).await
}
async fn scoped_stage_subjektiv_candidate(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -21529,14 +21466,9 @@ async fn spawn_known_worker(
operation_id: scoped_worker_control_operation_id(&controller, &operation_id),
input_fingerprint,
});
let response = create_workspace_worker_inner(
api.clone(),
headers,
request,
resolved_control_operation,
false,
)
.await?;
let response =
create_workspace_worker_inner(api.clone(), headers, request, resolved_control_operation)
.await?;
if let Err(error) = api
.store
.create_worker_control_grant(&WorkerControlGrantRecord {
@@ -27110,7 +27042,7 @@ async fn create_workspace_worker(
headers: HeaderMap,
Json(request): Json<CreateWorkspaceWorkerRequest>,
) -> ApiResult<Json<BrowserCreateWorkerResponse>> {
create_workspace_worker_inner(api, headers, request, None, false).await
create_workspace_worker_inner(api, headers, request, None).await
}
async fn create_workspace_worker_inner(
@@ -27118,33 +27050,19 @@ async fn create_workspace_worker_inner(
headers: HeaderMap,
request: CreateWorkspaceWorkerRequest,
resolved_control_operation: Option<WorkerControlOperation>,
resolved_subjektiv_attached: bool,
) -> ApiResult<Json<BrowserCreateWorkerResponse>> {
let CreateWorkspaceWorkerRequest {
runtime_id,
display_name,
singleton_key,
mut singleton_key,
profile,
ticket_assignment,
initial_submit,
workdir_attachments,
feature_connections,
control_operation_id: _,
} = request;
if resolved_subjektiv_attached {
let singleton_key = singleton_key.as_deref().ok_or_else(|| {
Error::InvalidInput(
"subject-attached Worker launch requires a Backend-owned singleton key".to_string(),
)
})?;
if !singleton_key.starts_with(SUBJEKTIV_SINGLETON_PREFIX) {
return Err(Error::InvalidInput(
"subject-attached Worker launch requires a subject singleton key".to_string(),
)
.into());
}
} else {
validate_caller_worker_singleton_key(singleton_key.as_deref())?;
}
let source = optional_worker_mutation_source(&api, &api.config.workspace_id, &headers)?;
let config_state = api
.config_store
.load_workspace_config(&api.config.workspace_id)?
@@ -27185,6 +27103,34 @@ async fn create_workspace_worker_inner(
&profile,
prompt_catalog.as_ref(),
)?;
let resolved_subjektiv_attached = match feature_connections.subjektiv {
None => {
validate_caller_worker_singleton_key(singleton_key.as_deref())?;
false
}
Some(connection) => {
if singleton_key.is_some() {
return Err(Error::InvalidInput(
"singleton_key must be omitted when a subjektiv subject connection is requested"
.to_string(),
)
.into());
}
let profile_manifest =
crate::profile_settings::resolve_profile_manifest_from_config_bundle(
resolved_config_bundle.as_ref().ok_or_else(|| {
Error::Config("resolved Worker Profile bundle is missing".to_string())
})?,
&profile,
)?;
let attachment = open_subjektiv_store(&api)?
.authorize_worker_attachment(&connection.subject_id, &profile_manifest)
.map_err(subjektiv_store_error)?;
crate::store::validate_worker_singleton_key(&attachment.singleton_key)?;
singleton_key = Some(attachment.singleton_key);
true
}
};
let display_name = sanitize_worker_display_name(&display_name).ok_or_else(|| {
settings_bad_request(
"invalid_worker_display_name",
@@ -27195,7 +27141,6 @@ async fn create_workspace_worker_inner(
return Err(Error::ReservedWorkerName(display_name).into());
}
validate_worker_initial_submit(&initial_submit)?;
let source = optional_worker_mutation_source(&api, &api.config.workspace_id, &headers)?;
reject_orchestrator_generic_flow_spawn(
&api,
source.as_ref(),
@@ -29771,32 +29716,55 @@ fn worker_launch_options_response(api: &WorkspaceApi) -> ApiResult<WorkerLaunchO
.ok_or_else(|| {
ApiError::from(Error::InvalidRecordId("virtual config source tree".into()))
})?;
let profile_settings = crate::profile_settings::project_profiles_from_workspace_config(
let profile_projection = crate::profile_settings::project_profiles_from_workspace_config(
&api.config.workspace_id,
&config_state,
)?
.settings;
let profiles = profile_settings
)?;
let prompt_catalog = api
.prompt_projection_cache
.resolve(&api.config.workspace_id, &config_state)?;
let profiles = profile_projection
.settings
.profiles
.into_iter()
.iter()
.filter(|profile| {
!profile
.diagnostics
.iter()
.any(|diagnostic| diagnostic.severity == server_api::DiagnosticSeverity::Error)
})
.map(|profile| WorkerLaunchProfileCandidate {
id: profile.profile_id,
label: profile.label,
description: profile
.description
.unwrap_or_else(|| "Workspace profile.".to_string()),
.map(|profile| {
let bundle =
crate::profile_settings::build_virtual_profile_config_bundle_with_prompt_projection(
&profile_projection,
&config_state,
&api.config.workspace_id,
&api.config.workspace_created_at,
&profile.profile_id,
prompt_catalog.as_ref(),
)?
.ok_or_else(|| Error::Config("resolved Worker Profile bundle is missing".into()))?;
let manifest = crate::profile_settings::resolve_profile_manifest_from_config_bundle(
&bundle,
&profile.profile_id,
)?;
Ok(WorkerLaunchProfileCandidate {
id: profile.profile_id.clone(),
label: profile.label.clone(),
description: profile
.description
.clone()
.unwrap_or_else(|| "Workspace profile.".to_string()),
feature_connections: server_api::WorkerLaunchFeatureConnectionOptions {
subjektiv: manifest.feature.subjektiv.profile.enabled,
},
})
})
.collect();
.collect::<ApiResult<Vec<_>>>()?;
Ok(WorkerLaunchOptionsResponse {
workspace_id: api.config.workspace_id.clone(),
runtimes,
default_profile: profile_settings.default_profile,
default_profile: profile_projection.settings.default_profile.clone(),
profiles,
repositories: working_directory_repository_options(api),
working_directories: available_working_directory_summaries(api)
@@ -32729,6 +32697,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -33258,6 +33227,15 @@ mod tests {
}
let options = worker_launch_options_response(&api).unwrap();
assert!(
options
.profiles
.iter()
.find(|profile| profile.id == "builtin:companion")
.unwrap()
.feature_connections
.subjektiv
);
let available = options
.working_directories
.iter()
@@ -33306,6 +33284,7 @@ mod tests {
relative_cwd: None,
},
],
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -33543,6 +33522,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36389,6 +36369,7 @@ mod tests {
selector: "builtin:coder-review".to_string(),
}],
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36423,6 +36404,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36458,6 +36440,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36547,6 +36530,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36805,6 +36789,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -36864,6 +36849,7 @@ mod tests {
"working_directory_id": fixture.read_only_workdir_id,
}
],
"feature_connections": {},
"control_operation_id": operation_id,
})
}
@@ -37287,6 +37273,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -37328,6 +37315,7 @@ mod tests {
"working_directory_id": read_only_workdir_id,
}
],
"feature_connections": {},
"control_operation_id": "guarded-strict-remote-success",
});
assert!(payload.get("working_directory").is_none());
@@ -37580,6 +37568,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -37612,6 +37601,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: Some("control-spawn-retry".to_string()),
};
@@ -37687,6 +37677,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
};
@@ -37845,6 +37836,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -37943,6 +37935,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
};
let Json(controller) = create_workspace_worker(
@@ -38555,6 +38548,15 @@ mod tests {
);
let memory_root = format!("{subject_root}/memories/{}", memory.id);
request_json_authenticated(
app.clone(),
"POST",
&format!("{subject_root}/worker"),
Some(serde_json::json!({})),
&owner_token,
StatusCode::METHOD_NOT_ALLOWED,
)
.await;
request_json(
app.clone(),
"GET",
@@ -38915,6 +38917,107 @@ mod tests {
assert_eq!(snapshot.memory_refs[0].revision, expected_ref.revision);
}
#[tokio::test]
async fn ordinary_worker_launch_validates_optional_subjektiv_connection() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let store = open_subjektiv_store(&api).unwrap();
let active = store
.create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
.unwrap();
let retired = store
.create_subject(crate::subjektiv::SubjectRole::new("retired").unwrap())
.unwrap();
store.retire_subject(&retired.id).unwrap();
let request = |subject_id: Option<&str>, profile: &str, singleton_key: Option<&str>| {
CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Ordinary launch".to_string(),
singleton_key: singleton_key.map(ToOwned::to_owned),
profile: Some(profile.to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: subject_id.map(|subject_id| {
server_api::SubjektivWorkerConnectionRequest {
subject_id: subject_id.to_string(),
}
}),
},
control_operation_id: None,
}
};
let Json(ordinary) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(request(None, "builtin:companion", None)),
)
.await
.unwrap();
assert!(ordinary.worker.singleton_key.is_none());
for (candidate, profile, singleton_key, expected_status) in [
(
"missing-subject",
"builtin:companion",
None,
StatusCode::NOT_FOUND,
),
(
retired.id.as_str(),
"builtin:companion",
None,
StatusCode::BAD_REQUEST,
),
(
active.id.as_str(),
"builtin:standalone",
None,
StatusCode::BAD_REQUEST,
),
(
active.id.as_str(),
"builtin:companion",
Some("caller-key"),
StatusCode::BAD_REQUEST,
),
] {
let response = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(request(Some(candidate), profile, singleton_key)),
)
.await
.unwrap_err()
.into_response();
assert_eq!(response.status(), expected_status, "candidate={candidate}");
}
let Json(first) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(request(Some(&active.id), "builtin:companion", None)),
)
.await
.unwrap();
let Json(retried) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(request(Some(&active.id), "builtin:companion", None)),
)
.await
.unwrap();
assert_eq!(retried.worker_id, first.worker_id);
assert_eq!(
first.worker.singleton_key.as_deref(),
Some(format!("subjektiv:{}", active.id).as_str())
);
}
#[tokio::test]
async fn subjektiv_subject_worker_and_staging_derive_scope_from_singleton_authority() {
let workspace = tempfile::tempdir().unwrap();
@@ -38935,23 +39038,25 @@ mod tests {
.unwrap()
.create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap())
.unwrap();
let request = server_api::SubjektivSubjectWorkerStartRequest {
worker: CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Subject Worker".to_string(),
singleton_key: None,
profile: Some("builtin:companion".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
control_operation_id: None,
let request = CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Subject Worker".to_string(),
singleton_key: None,
profile: Some("builtin:companion".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
subject_id: subject.id.clone(),
}),
},
control_operation_id: None,
};
let Json(created) = scoped_start_subjektiv_subject_worker(
let Json(created) = scoped_create_workspace_worker(
State(api.clone()),
AxumPath(ScopedSubjektivSubjectPath {
AxumPath(ScopedWorkspacePath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
subject_id: subject.id.clone(),
}),
HeaderMap::new(),
Json(request),
@@ -40259,6 +40364,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -40278,6 +40384,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -40362,6 +40469,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -45168,6 +45276,7 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
}),
)
@@ -51507,7 +51616,8 @@ mod tests {
"alias": "workdir",
"working_directory_id": working_directory_id,
"relative_cwd": "../escape"
}]
}],
"feature_connections": {}
})),
StatusCode::BAD_REQUEST,
)
@@ -52207,7 +52317,8 @@ mod tests {
serde_json::json!({
"runtime_id": "embedded-worker-runtime",
"display_name": "",
"initial_submit": []
"initial_submit": [],
"feature_connections": {}
}),
)
.await;
@@ -52326,10 +52437,10 @@ mod tests {
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
},
None,
false,
)
.await
.unwrap();
@@ -54746,7 +54857,7 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
assert!(subjektiv_scope_from_singleton_key("subjektiv:").is_err());
assert!(subjektiv_scope_from_singleton_key("subjektiv: subject-42").is_err());
assert_eq!(
subjektiv_singleton_key("subject-42").unwrap(),
crate::subjektiv::subject_worker_singleton_key("subject-42").unwrap(),
"subjektiv:subject-42"
);
}
crates/workspace-server/src/subjektiv.rs
+42
View File
@@ -22,6 +22,7 @@ use crate::feature_storage::{
pub const SUBJEKTIV_SCHEMA_VERSION: u32 = 1;
pub const SUBJEKTIV_FEATURE_ID: &str = "subjektiv";
pub const SUBJECT_WORKER_SINGLETON_PREFIX: &str = "subjektiv:";
pub const MAX_STAGING_ANCHORS: usize = 10;
/// Deterministic surface-generation policy. Input estimates use the repository's
/// provider-independent UTF-8 byte estimate (`ceil(bytes / 4)`). The material
@@ -45,6 +46,8 @@ pub enum SubjektivError {
SubjectNotFound(String),
#[error("subject `{0}` is retired")]
SubjectRetired(String),
#[error("the selected Worker Profile does not enable subjektiv")]
ProfileDisabled,
#[error("staging candidate `{0}` was not found for this subject")]
CandidateNotFound(String),
#[error("staging candidate `{0}` already exists with different content")]
@@ -147,6 +150,23 @@ pub struct SubjectPage {
pub has_more: bool,
}
/// Backend-authorized launch material for one active subject. The singleton key
/// is derived by subjektiv and then acquired through Yoi's generic singleton path.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthorizedSubjectWorkerAttachment {
pub singleton_key: String,
}
pub fn subject_worker_singleton_key(subject_id: &str) -> Result<String> {
validate_label("subject id", subject_id)?;
if subject_id.trim() != subject_id {
return Err(SubjektivError::InvalidRecord(
"subject id must not contain surrounding whitespace".to_string(),
));
}
Ok(format!("{SUBJECT_WORKER_SINGLETON_PREFIX}{subject_id}"))
}
/// Immutable host-recorded history that attributes one durable Session to the
/// subject and Worker that produced it. This is archival provenance only: it is
/// deliberately not a current-Worker link or execution ownership record.
@@ -1091,6 +1111,28 @@ impl SubjektivStore {
})
}
/// Validates an explicit product launch connection against this Workspace's
/// subject store and the already-resolved Profile policy. Subject knowledge
/// alone is insufficient: only an active local record produces host launch material.
pub fn authorize_worker_attachment(
&self,
subject_id: &str,
profile: &manifest::WorkerManifest,
) -> Result<AuthorizedSubjectWorkerAttachment> {
if !profile.feature.subjektiv.profile.enabled {
return Err(SubjektivError::ProfileDisabled);
}
let subject = self
.subject(subject_id)?
.ok_or_else(|| SubjektivError::SubjectNotFound(subject_id.to_string()))?;
if subject.state != SubjectState::Active {
return Err(SubjektivError::SubjectRetired(subject_id.to_string()));
}
Ok(AuthorizedSubjectWorkerAttachment {
singleton_key: subject_worker_singleton_key(&subject.id)?,
})
}
/// Lists the first deterministic bounded page of subjects. Ordering uses
/// immutable creation time with the host-issued subject id as a total-order
/// tie breaker.