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

feat: integrate durable Ticket assignment tracking with current develop

Browse Source
This commit is contained in:
Hare
2026-10-12 03:26:15 +09:00
parent 27bf74a051 96e5ceb4fb
commit a6a97bca74
35 changed files with 7891 additions and 507 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/assignment_tracking.rs
+173
View File
@@ -0,0 +1,173 @@
//! Workspace adapter: Ticket owns the assignment/tracking schema and mutations.
//! Authenticated source is constructed by Backend after singleton resolution;
//! grants, work releases and startup recovery never derive from tracking status.
use super::*;
/// Backend-authenticated attribution. Readable assigned_by/actor is audit
/// payload, not this authority context. Never decoded from model/API input.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TicketOrchestratorAssignmentSourceContext {
Runtime {
runtime_id: String,
worker_id: String,
agent_key: String,
authentication_binding_id: String,
},
Account {
account_id: String,
user_id: String,
},
}
pub(super) fn require_runtime_source(
conn: &Connection,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
agent_key: &str,
binding_id: &str,
) -> Result<()> {
let current_source:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM worker_singleton_owners singleton JOIN workspace_runtime_bindings binding ON binding.workspace_id=singleton.workspace_id AND binding.runtime_id=singleton.runtime_id WHERE singleton.workspace_id=?1 AND singleton.singleton_key=?2 AND singleton.runtime_id=?3 AND singleton.worker_id=?4 AND binding.binding_id=?5 AND binding.revoked_at IS NULL)",params![workspace_id,crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY,runtime_id,worker_id,binding_id],|row|row.get(0))?;
if agent_key != "workspace-orchestrator" || binding_id.trim().is_empty() || !current_source {
return Err(Error::WorkspacePermissionDenied("Assignment source is not the current registered Orchestrator with its authenticated Runtime binding".into()));
}
Ok(())
}
pub(super) fn require_assignment_source(
conn: &Connection,
workspace_id: &str,
source: &TicketOrchestratorAssignmentSourceContext,
) -> Result<()> {
match source {
TicketOrchestratorAssignmentSourceContext::Runtime {
runtime_id,
worker_id,
agent_key,
authentication_binding_id,
} => require_runtime_source(
conn,
workspace_id,
runtime_id,
worker_id,
agent_key,
authentication_binding_id,
),
TicketOrchestratorAssignmentSourceContext::Account {
account_id,
user_id,
} => {
// Backend already authenticates Account/Workspace permission. Do not
// infer a model role or require a Runtime for account authoring.
let exists: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM accounts WHERE account_id=?1)",
[account_id],
|row| row.get(0),
)?;
if !exists || user_id.trim().is_empty() {
return Err(Error::WorkspacePermissionDenied(
"Assignment account source is not a valid Backend actor".into(),
));
}
Ok(())
}
}
}
pub(super) fn update(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
update: &TicketAssignmentTrackingUpdate,
source: &TicketAssignmentTrackingSourceContext,
) -> Result<TicketAssignmentTrackingProjection> {
store.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
// Every call, including a receipt replay, fences the authenticated
// Runtime binding and singleton snapshot under the write lock. Assignment
// replacement/removal may be replayed, Worker replacement may not.
require_runtime_source(
&tx,
workspace_id,
&source.runtime_id,
&source.worker_id,
&source.agent_key,
&source.authentication_binding_id,
)?;
let result = ticket::sqlite_update_ticket_assignment_tracking_in_transaction(
&tx,
workspace_id,
ticket_id,
update,
source,
)
.map_err(|error| Error::TicketAssignmentConflict(error.to_string()))?;
tx.commit()?;
Ok(result)
})
}
pub(super) fn get(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
) -> Result<Option<TicketAssignmentTrackingProjection>> {
store.with_conn(|conn| {
ticket::sqlite_ticket_assignment_tracking(conn, workspace_id, ticket_id, assignment_id)
.map_err(|error| Error::Store(error.to_string()))
})
}
pub(super) fn current(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
) -> Result<Vec<TicketAssignmentTrackingProjection>> {
store.with_conn(|conn| {
ticket::sqlite_current_ticket_assignment_tracking(conn, workspace_id, ticket_id)
.map_err(|error| Error::Store(error.to_string()))
})
}
pub(super) fn history(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
) -> Result<Vec<TicketAssignmentTrackingHistoryEntry>> {
store.with_conn(|conn| {
ticket::sqlite_ticket_assignment_tracking_history(
conn,
workspace_id,
ticket_id,
assignment_id,
)
.map_err(|error| Error::Store(error.to_string()))
})
}
pub(super) fn history_page(
store: &SqliteWorkspaceStore,
workspace_id: &str,
ticket_id: &str,
assignment_id: &str,
after_event_id: Option<&str>,
limit: usize,
) -> Result<ticket::TicketAssignmentTrackingHistoryPage> {
store.with_conn(|conn| {
ticket::sqlite_ticket_assignment_tracking_history_page(
conn,
workspace_id,
ticket_id,
assignment_id,
after_event_id,
limit,
)
.map_err(|error| match error {
ticket::TicketError::Conflict(message) => Error::InvalidInput(message),
other => Error::Store(other.to_string()),
})
})
}
crates/workspace-server/src/assignment_tracking_tests.rs
+1486
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/authority.rs
+19
View File
@@ -238,6 +238,20 @@ impl SqliteWorkspaceAuthority {
})
}
/// Assignment self-report enrichment is a separate projection so retained
/// responsibility and existing Ticket constructors stay distinct from status.
pub fn ticket_assignment_tracking(
&self,
reference: &str,
) -> Result<Vec<ticket::TicketAssignmentTrackingProjection>> {
let ticket = self
.ticket_backend
.show(TicketIdOrSlug::from(reference))
.map_err(Error::from)?;
self.store
.list_current_ticket_assignment_tracking(&self.workspace_id, &ticket.meta.id)
}
pub fn with_source_commit_resolver(
mut self,
source_commit_resolver: Arc<dyn TicketSourceCommitResolver>,
@@ -905,7 +919,11 @@ impl SqliteWorkspaceAuthority {
&blocker.blocking_ticket,
)?;
}
let assignment_tracking = self
.store
.list_current_ticket_assignment_tracking(&self.workspace_id, &ticket.meta.id)?;
Ok(TicketDetail {
assignment_tracking,
id: ticket.meta.id,
resource_key,
title: ticket.meta.title,
@@ -2330,6 +2348,7 @@ fn ticket_query_item(
}
}
TicketQueryItem {
assignment_tracking: detail.assignment_tracking.clone(),
id: summary.id,
resource_key: summary.resource_key,
title: summary.title,
crates/workspace-server/src/latest_schema.sql
+10 -104
View File
@@ -328,31 +328,7 @@ CREATE TABLE server_secret_objects (
REFERENCES repository_ssh_credential_keys(workspace_id, credential_id, operation_id)
ON DELETE CASCADE
);
CREATE TABLE ticket_assignment_operations (
workspace_id TEXT NOT NULL,
operation_id TEXT NOT NULL,
action TEXT NOT NULL CHECK (action IN ('assign', 'reassign', 'unassign')),
ticket_id TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'worker'
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
principal_kind TEXT NOT NULL DEFAULT 'worker'
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
principal_id TEXT,
runtime_id TEXT,
worker_id TEXT,
assignment_id TEXT,
expected_assignment_id TEXT,
created_at TEXT NOT NULL,
request_fingerprint TEXT,
claim_state TEXT NOT NULL DEFAULT 'committed'
CHECK(claim_state IN ('pending', 'failed', 'committed')),
failure_reason TEXT,
binding_recovery_json TEXT,
PRIMARY KEY (workspace_id, operation_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
CHECK(claim_state = 'committed' OR assignment_id IS NULL),
CHECK((claim_state = 'failed') = (failure_reason IS NOT NULL))
);
-- Ticket-owned assignment schema is composed by store.rs.
CREATE TABLE ticket_assignment_ticket_tombstones (
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
@@ -368,72 +344,9 @@ CREATE TABLE ticket_assignment_worker_tombstones (
PRIMARY KEY (workspace_id, runtime_id, worker_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
);
CREATE TABLE ticket_current_worker_assignments (
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'worker'
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
assignment_id TEXT NOT NULL,
principal_kind TEXT NOT NULL DEFAULT 'worker'
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
principal_id TEXT,
runtime_id TEXT,
worker_id TEXT,
updated_at TEXT NOT NULL,
PRIMARY KEY (workspace_id, ticket_id, role, assignment_id),
UNIQUE (workspace_id, assignment_id),
FOREIGN KEY (workspace_id, ticket_id, role, assignment_id)
REFERENCES ticket_worker_assignments(workspace_id, ticket_id, role, assignment_id)
ON DELETE CASCADE,
FOREIGN KEY (workspace_id, ticket_id)
REFERENCES typed_tickets(workspace_id, ticket_id)
ON DELETE CASCADE,
CHECK(
(principal_kind = 'worker' AND runtime_id IS NOT NULL AND worker_id IS NOT NULL AND principal_id IS NULL)
OR
(principal_kind != 'worker' AND runtime_id IS NULL AND worker_id IS NULL AND principal_id IS NOT NULL AND length(trim(principal_id)) > 0)
)
);
CREATE TABLE ticket_worker_assignment_events (
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'worker'
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
event_id TEXT NOT NULL,
action TEXT NOT NULL CHECK (action IN ('assigned', 'reassigned', 'unassigned')),
assignment_id TEXT,
previous_assignment_id TEXT,
actor TEXT NOT NULL,
created_at TEXT NOT NULL,
operation_id TEXT,
reason TEXT,
PRIMARY KEY (workspace_id, event_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
);
CREATE TABLE ticket_worker_assignments (
workspace_id TEXT NOT NULL,
ticket_id TEXT NOT NULL,
assignment_id TEXT NOT NULL,
role TEXT NOT NULL DEFAULT 'worker'
CHECK(role IN ('orchestrator', 'worker', 'owner', 'contributor')),
principal_kind TEXT NOT NULL DEFAULT 'worker'
CHECK(principal_kind IN ('user', 'worker', 'workspace_agent')),
principal_id TEXT,
runtime_id TEXT,
worker_id TEXT,
assigned_by TEXT NOT NULL,
assigned_at TEXT NOT NULL,
PRIMARY KEY (workspace_id, assignment_id),
UNIQUE (workspace_id, ticket_id, assignment_id),
UNIQUE (workspace_id, ticket_id, role, assignment_id),
UNIQUE (workspace_id, ticket_id, role, assignment_id, principal_kind, principal_id, runtime_id, worker_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
CHECK(
(principal_kind = 'worker' AND runtime_id IS NOT NULL AND worker_id IS NOT NULL AND principal_id IS NULL)
OR
(principal_kind != 'worker' AND runtime_id IS NULL AND worker_id IS NULL AND principal_id IS NOT NULL AND length(trim(principal_id)) > 0)
)
);
CREATE TABLE workspace_runtime_bindings (
workspace_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
@@ -1165,14 +1078,10 @@ CREATE INDEX idx_repository_ssh_credentials_workspace_status
ON repository_ssh_credentials(workspace_id, status, credential_id);
CREATE INDEX idx_repository_ssh_host_trusts_workspace_host
ON repository_ssh_host_trusts(workspace_id, hostname, port);
CREATE INDEX idx_ticket_assignment_operations_ticket
ON ticket_assignment_operations(workspace_id, ticket_id, role, created_at DESC);
CREATE INDEX idx_ticket_worker_assignment_events_ticket
ON ticket_worker_assignment_events(workspace_id, ticket_id, role, created_at DESC);
CREATE INDEX idx_ticket_worker_assignments_principal
ON ticket_worker_assignments(workspace_id, role, principal_kind, principal_id, runtime_id, worker_id, assigned_at DESC);
CREATE INDEX idx_ticket_worker_assignments_ticket
ON ticket_worker_assignments(workspace_id, ticket_id, role, assigned_at DESC);
CREATE INDEX idx_workspace_runtime_bindings_workspace
ON workspace_runtime_bindings(workspace_id, revoked_at, runtime_id);
CREATE INDEX idx_typed_ticket_relations_workspace_target
@@ -1202,11 +1111,8 @@ CREATE INDEX idx_workspace_resource_keys_reverse
ON workspace_resource_keys(workspace_id, resource_kind, resource_key);
CREATE INDEX repositories_workspace_provider_idx
ON repositories(workspace_id, provider);
CREATE INDEX ticket_current_principal_idx
ON ticket_current_worker_assignments(workspace_id, principal_kind, principal_id, runtime_id, worker_id);
CREATE UNIQUE INDEX ticket_current_singleton_role_idx
ON ticket_current_worker_assignments(workspace_id, ticket_id, role)
WHERE role IN ('orchestrator', 'worker');
CREATE INDEX typed_ticket_events_workspace_kind_ticket
ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id
crates/workspace-server/src/server.rs
+355 -51
View File
@@ -2808,7 +2808,7 @@ async fn authorize_workspace_api_request(
next: axum::middleware::Next,
) -> Response {
if !request.uri().path().starts_with("/api/")
&& ticket_case_workspace_id(request.uri().path()).is_none()
&& canonical_ticket_operation_workspace_id(request.uri().path()).is_none()
{
return next.run(request).await;
}
@@ -3031,7 +3031,7 @@ async fn dispatch_workspace_request(
if let Some(workspace_id) = workspace_id
&& (path.starts_with("/api/w/")
|| path.starts_with("/api/runtime/v1/workspaces/")
|| ticket_case_workspace_id(&path).is_some())
|| canonical_ticket_operation_workspace_id(&path).is_some())
&& let Err(response) =
authorize_scoped_workspace_request(&api, workspace_id, &mut request).await
{
@@ -3105,9 +3105,9 @@ async fn serve_server_static_shell(api: &WorkspaceServerApi, path: &str) -> Resp
static_file_or_spa_response(static_root, path).await
}
// The canonical case route is deliberately not an /api/w alias. Recognize only
// this operation so its authentication/scope is identical without broadening CRUD.
fn ticket_case_workspace_id(path: &str) -> Option<&str> {
// Canonical case/tracking routes are deliberately not /api/w aliases. Recognize
// only these operations so authentication/scope are identical without exposing CRUD.
fn canonical_ticket_operation_workspace_id(path: &str) -> Option<&str> {
let mut segments = path.strip_prefix("/workspaces/")?.split('/');
let workspace_id = segments.next()?;
let resource = segments.next()?;
@@ -3116,13 +3116,20 @@ fn ticket_case_workspace_id(path: &str) -> Option<&str> {
(!workspace_id.is_empty()
&& resource == "tickets"
&& !ticket_id.is_empty()
&& operation == "case"
&& segments.next().is_none())
&& match operation {
"case" => segments.next().is_none(),
"assignment-tracking" => match segments.next() {
None => true,
Some("history") => segments.next().is_none(),
_ => false,
},
_ => false,
})
.then_some(workspace_id)
}
fn scoped_workspace_id(path: &str) -> Option<&str> {
if let Some(workspace_id) = ticket_case_workspace_id(path) {
if let Some(workspace_id) = canonical_ticket_operation_workspace_id(path) {
return Some(workspace_id);
}
if let Some(rest) = path.strip_prefix("/api/runtime/v1/workspaces/") {
@@ -6232,6 +6239,7 @@ async fn attach_server_request_context(
.extensions()
.get::<crate::worker_source::VerifiedRuntimeRequestSource>()
.map(|source| server_api::ServerRuntimeSource {
authentication_binding_id: source.authentication_binding_id.clone(),
runtime_id: source.runtime_id.clone(),
worker_id: source.worker_id.clone(),
});
@@ -6601,6 +6609,11 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::ticket_case_update(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignment_tracking_get(
service.clone(),
))
.merge(server_api::server_api_axum::ticket_assignment_tracking_update(service.clone()))
.merge(server_api::server_api_axum::ticket_assignment_tracking_history(service.clone()))
.merge(server_api::server_api_axum::workspace_ticket_tag_list(
service.clone(),
))
@@ -10538,6 +10551,75 @@ impl server_api::ServerApi for ServerApiContractService {
Ok(response)
}
async fn ticket_assignment_tracking_get(
&self,
_context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentTrackingQuery,
) -> std::result::Result<server_api::TicketAssignmentTrackingList, server_api::RepositoryApiError>
{
let api = self.workspace_api()?;
validate_workspace_scope(api, &workspace_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let ticket_id = resolve_workspace_ticket_reference(api, &workspace_id, &ticket_id)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let values = if let Some(assignment_id) = query.assignment_id {
api.store
.get_ticket_assignment_tracking(&workspace_id, &ticket_id, &assignment_id)
.map(|value| value.into_iter().collect::<Vec<_>>())
} else {
api.authority.ticket_assignment_tracking(&ticket_id)
}
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
Ok(server_api::TicketAssignmentTrackingList(values))
}
async fn ticket_assignment_tracking_update(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
request: server_api::TicketAssignmentTrackingUpdateRequest,
) -> std::result::Result<
server_api::TicketAssignmentTrackingRecord,
server_api::RepositoryApiError,
> {
let Json(value) = scoped_ticket_assignment_tracking_update_with_context(
State(self.workspace_api()?.clone()),
contract_request_headers(&context)?,
&context,
AxumPath((workspace_id, ticket_id)),
Json(request.0),
)
.await
.map_err(ApiError::into_repository_api_error)?;
Ok(server_api::TicketAssignmentTrackingRecord(value))
}
async fn ticket_assignment_tracking_history(
&self,
_context: server_api::ServerRequestContext,
workspace_id: String,
ticket_id: String,
query: server_api::TicketAssignmentTrackingHistoryQuery,
) -> std::result::Result<
server_api::TicketAssignmentTrackingHistoryPage,
server_api::RepositoryApiError,
> {
let Json(value) = scoped_ticket_assignment_tracking_history(
State(self.workspace_api()?.clone()),
AxumPath((workspace_id, ticket_id)),
Query(query),
)
.await
.map_err(ApiError::into_repository_api_error)?;
Ok(value)
}
async fn ticket_assignment_list(
&self,
workspace_id: String,
@@ -10557,6 +10639,7 @@ impl server_api::ServerApi for ServerApiContractService {
async fn ticket_assignment_set(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
id: String,
role: String,
@@ -10565,8 +10648,10 @@ impl server_api::ServerApi for ServerApiContractService {
server_api::TicketRoleAssignmentMutationResponse,
server_api::RepositoryApiError,
> {
let Json(response) = scoped_set_ticket_assignment(
let Json(response) = scoped_set_ticket_assignment_with_context(
State(self.workspace_api()?.clone()),
contract_request_headers(&context)?,
&context,
AxumPath((workspace_id, id, role)),
Json(request),
)
@@ -10577,6 +10662,7 @@ impl server_api::ServerApi for ServerApiContractService {
async fn ticket_assignment_clear(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
id: String,
role: String,
@@ -10585,8 +10671,10 @@ impl server_api::ServerApi for ServerApiContractService {
server_api::TicketRoleAssignmentMutationResponse,
server_api::RepositoryApiError,
> {
let Json(response) = scoped_clear_ticket_assignment(
let Json(response) = scoped_clear_ticket_assignment_with_context(
State(self.workspace_api()?.clone()),
contract_request_headers(&context)?,
&context,
AxumPath((workspace_id, id, role)),
Query(query),
)
@@ -13667,10 +13755,85 @@ async fn finalize_manual_ticket_worker_claim(
}
}
#[cfg(test)]
async fn scoped_set_ticket_assignment(
state: State<WorkspaceApi>,
path: AxumPath<(String, String, String)>,
request: Json<server_api::SetTicketRoleAssignmentRequest>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
apply_set_ticket_assignment(state, path, request, "workspace-web".into(), None).await
}
fn authenticated_assignment_source(
api: &WorkspaceApi,
headers: &HeaderMap,
context: &server_api::ServerRequestContext,
) -> Result<(
String,
crate::store::TicketOrchestratorAssignmentSourceContext,
)> {
if let Some(source) = authenticated_ticket_worker_source(headers, context)? {
if !find_workspace_orchestrator(api).is_some_and(|worker| worker.worker == source) {
return Err(Error::WorkspacePermissionDenied(
"Ticket assignment mutations require the registered current Workspace Orchestrator"
.into(),
));
}
let authentication_binding_id = context
.runtime_source
.as_ref()
.map(|source| source.authentication_binding_id.clone())
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Assignment requires the Runtime binding used to authenticate this Worker"
.into(),
)
})?;
let actor = format!("worker:{}/{}", source.runtime_id, source.worker_id);
return Ok((
actor,
crate::store::TicketOrchestratorAssignmentSourceContext::Runtime {
runtime_id: source.runtime_id,
worker_id: source.worker_id,
agent_key: "workspace-orchestrator".into(),
authentication_binding_id,
},
));
}
let actor = context.actor.as_ref().ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Ticket assignment mutations require an authenticated actor".into(),
)
})?;
let identity = serde_json::to_string(&(&actor.account_id, &actor.user_id))
.map_err(|error| Error::Config(format!("failed to encode assignment actor: {error}")))?;
Ok((
format!("workspace-user:{identity}"),
crate::store::TicketOrchestratorAssignmentSourceContext::Account {
account_id: actor.account_id.clone(),
user_id: actor.user_id.clone(),
},
))
}
async fn scoped_set_ticket_assignment_with_context(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
context: &server_api::ServerRequestContext,
path: AxumPath<(String, String, String)>,
request: Json<server_api::SetTicketRoleAssignmentRequest>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
validate_workspace_scope(&api, &path.0.0)?;
let (author, source) = authenticated_assignment_source(&api, &headers, context)?;
apply_set_ticket_assignment(State(api), path, request, author, Some(source)).await
}
async fn apply_set_ticket_assignment(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, id, role)): AxumPath<(String, String, String)>,
Json(request): Json<server_api::SetTicketRoleAssignmentRequest>,
assigned_by: String,
source: Option<crate::store::TicketOrchestratorAssignmentSourceContext>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
validate_workspace_scope(&api, &workspace_id)?;
let ticket = api.authority.ticket(&id)?;
@@ -13688,33 +13851,30 @@ async fn scoped_set_ticket_assignment(
assignment_id: new_id("tasg"),
role,
principal,
assigned_by: "workspace-web".to_string(),
assigned_by,
assigned_at,
};
let assignment = match role {
TicketAssignmentRole::Orchestrator => {
if api
.store
.get_active_ticket_role_assignment(
&workspace_id,
&ticket.id,
TicketAssignmentRole::Worker,
if let Some(source) = source.as_ref() {
api.store.set_current_ticket_orchestrator_assignment(
&record,
expected_assignment_id.as_deref(),
&new_id("tasev"),
&operation_id,
expected_assignment_id.is_some(),
source,
)?
} else {
// Only the cfg(test) direct-handler adapter lacks a Backend source.
api.store.set_current_ticket_role_assignment(
&record,
expected_assignment_id.as_deref(),
&new_id("tasev"),
&operation_id,
expected_assignment_id.is_some(),
)?
.is_some()
{
return Err(Error::TicketAssignmentConflict(
"Orchestrator assignment conflicts with an active Worker assignment"
.to_string(),
)
.into());
}
api.store.set_current_ticket_role_assignment(
&record,
expected_assignment_id.as_deref(),
&new_id("tasev"),
&operation_id,
expected_assignment_id.is_some(),
)?
}
TicketAssignmentRole::Worker => {
if expected_assignment_id.is_some() {
@@ -13747,10 +13907,33 @@ async fn scoped_set_ticket_assignment(
}))
}
#[cfg(test)]
async fn scoped_clear_ticket_assignment(
state: State<WorkspaceApi>,
path: AxumPath<(String, String, String)>,
query: Query<server_api::ClearTicketRoleAssignmentQuery>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
apply_clear_ticket_assignment(state, path, query, "workspace-web".into(), None).await
}
async fn scoped_clear_ticket_assignment_with_context(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
context: &server_api::ServerRequestContext,
path: AxumPath<(String, String, String)>,
query: Query<server_api::ClearTicketRoleAssignmentQuery>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
validate_workspace_scope(&api, &path.0.0)?;
let (author, source) = authenticated_assignment_source(&api, &headers, context)?;
apply_clear_ticket_assignment(State(api), path, query, author, Some(source)).await
}
async fn apply_clear_ticket_assignment(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, id, role)): AxumPath<(String, String, String)>,
Query(query): Query<server_api::ClearTicketRoleAssignmentQuery>,
author: String,
source: Option<crate::store::TicketOrchestratorAssignmentSourceContext>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
validate_workspace_scope(&api, &workspace_id)?;
let ticket = api.authority.ticket(&id)?;
@@ -13769,17 +13952,36 @@ async fn scoped_clear_ticket_assignment(
.ok_or_else(|| {
Error::TicketAssignmentConflict("unassign requires assignment_id".to_string())
})?;
let cleared = api.store.clear_current_ticket_role_assignment(
&workspace_id,
&ticket.id,
role,
&assignment_id,
&new_id("tasev"),
&operation_id,
"workspace-web",
&Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
Some("role assignment removed from Ticket detail"),
)?;
let event_id = new_id("tasev");
let occurred_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
let reason = Some("role assignment removed from Ticket detail");
let cleared = if role == TicketAssignmentRole::Orchestrator
&& let Some(source) = source.as_ref()
{
api.store.clear_current_ticket_orchestrator_assignment(
&workspace_id,
&ticket.id,
&assignment_id,
&event_id,
&operation_id,
&author,
&occurred_at,
reason,
source,
)?
} else {
api.store.clear_current_ticket_role_assignment(
&workspace_id,
&ticket.id,
role,
&assignment_id,
&event_id,
&operation_id,
&author,
&occurred_at,
reason,
)?
};
if !cleared {
return Err(Error::TicketAssignmentConflict(format!(
"assignment `{assignment_id}` is not current for role `{}`",
@@ -16512,6 +16714,82 @@ async fn scoped_ticket_case_update(
apply_ticket_case_update(State(api), source, None, path, request).await
}
async fn scoped_ticket_assignment_tracking_update_with_context(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
context: &server_api::ServerRequestContext,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Json(update): Json<ticket::TicketAssignmentTrackingUpdate>,
) -> ApiResult<Json<ticket::TicketAssignmentTrackingProjection>> {
validate_workspace_scope(&api, &workspace_id)?;
let ticket_id = resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
let source = authenticated_ticket_worker_source(&headers, context)?.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Tracking is self-reported by a verified Runtime Worker".into(),
)
})?;
// Tracking authority is retained responsibility plus the registered singleton,
// NOT Ticket openness, active work, Worker state or a model-supplied role.
if !find_workspace_orchestrator(&api).is_some_and(|worker| worker.worker == source) {
return Err(Error::WorkspacePermissionDenied(
"Tracking updates require the registered current Workspace Orchestrator".into(),
)
.into());
}
let authentication_binding_id = context
.runtime_source
.as_ref()
.map(|source| source.authentication_binding_id.clone())
.ok_or_else(|| {
Error::WorkspacePermissionDenied(
"Tracking requires the Runtime binding actually used to authenticate the request"
.into(),
)
})?;
let source_context = ticket::TicketAssignmentTrackingSourceContext {
actor: format!("worker:{}/{}", source.runtime_id, source.worker_id),
runtime_id: source.runtime_id,
worker_id: source.worker_id,
agent_key: "workspace-orchestrator".into(),
authentication_binding_id,
};
// The Ticket-owned mutation checks current assignment/CAS and exact replay in
// one transaction. A historical receipt never restores operational grants.
let value = api.store.update_ticket_assignment_tracking(
&workspace_id,
&ticket_id,
&update,
&source_context,
)?;
Ok(Json(value))
}
async fn scoped_ticket_assignment_tracking_history(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Query(query): Query<server_api::TicketAssignmentTrackingHistoryQuery>,
) -> ApiResult<Json<server_api::TicketAssignmentTrackingHistoryPage>> {
validate_workspace_scope(&api, &workspace_id)?;
let ticket_id = resolve_workspace_ticket_reference(&api, &workspace_id, &ticket_id)?;
let limit = query.limit.unwrap_or(50);
if !(1..=100).contains(&limit) {
return Err(
Error::InvalidInput("tracking history limit must be between 1 and 100".into()).into(),
);
}
let page = api.store.page_ticket_assignment_tracking_history(
&workspace_id,
&ticket_id,
&query.assignment_id,
query.after_event_id.as_deref(),
limit,
)?;
Ok(Json(server_api::TicketAssignmentTrackingHistoryPage {
items: page.items,
next_event_id: page.next_event_id,
}))
}
async fn scoped_ticket_case_update_with_context(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
@@ -16519,6 +16797,20 @@ async fn scoped_ticket_case_update_with_context(
path: AxumPath<(String, String)>,
request: Json<ticket::TicketCaseUpdate>,
) -> ApiResult<Json<ticket::Ticket>> {
let source = authenticated_ticket_worker_source(&headers, context)?;
if source.is_none() && context.actor.is_none() {
return Err(Error::WorkspacePermissionDenied(
"Ticket case decisions require an authenticated actor".into(),
)
.into());
}
apply_ticket_case_update(State(api), source, context.actor.as_ref(), path, request).await
}
fn authenticated_ticket_worker_source(
headers: &HeaderMap,
context: &server_api::ServerRequestContext,
) -> Result<Option<WorkerMutationSource>> {
// Authentication already verified membership and the request's signature/body.
// Hints are redundant assertions, never authority or a required identity source.
let source = if let Some(source) = context.runtime_source.as_ref() {
@@ -16554,13 +16846,7 @@ async fn scoped_ticket_case_update_with_context(
.into());
}
}
if source.is_none() && context.actor.is_none() {
return Err(Error::WorkspacePermissionDenied(
"Ticket case decisions require an authenticated actor".into(),
)
.into());
}
apply_ticket_case_update(State(api), source, context.actor.as_ref(), path, request).await
Ok(source)
}
async fn apply_ticket_case_update(
@@ -28090,6 +28376,7 @@ async fn runtime_resource_fetch_contract(
message: "Runtime request proof is invalid".to_owned(),
})?;
server_api::ServerRuntimeSource {
authentication_binding_id: source.authentication_binding_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
}
@@ -32442,6 +32729,10 @@ impl IntoResponse for ApiError {
#[cfg(test)]
mod tests {
mod assignment_tracking_tests {
include!("server_assignment_tracking_tests.rs");
}
mod ticket_case_signed_tests {
include!("server_ticket_case_signed_tests.rs");
}
@@ -37892,6 +38183,7 @@ mod tests {
.uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"))
// Unit fixture stands below proof middleware, so inject its verified subject.
.extension(crate::worker_source::VerifiedRuntimeRequestSource {
authentication_binding_id: "fixture-binding".into(),
token_id_hash: None,
runtime_id: fixture.controller.runtime_id.clone(),
worker_id: Some(fixture.controller.worker_id.clone()),
@@ -38216,6 +38508,7 @@ mod tests {
.method("POST")
.uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"))
.extension(crate::worker_source::VerifiedRuntimeRequestSource {
authentication_binding_id: "fixture-binding".into(),
token_id_hash: None,
runtime_id: controller.runtime_id.clone(),
worker_id: Some(controller.worker_id.clone()),
@@ -40250,6 +40543,7 @@ mod tests {
actor: None,
worker_source: None,
runtime_source: Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: created.runtime_id.clone(),
worker_id: Some(created.worker_id.clone()),
}),
@@ -47969,8 +48263,10 @@ mod tests {
input.workflow_state = Some(TicketState::Open);
set_test_ticket_target(&mut input, "test-repository", "develop");
let ticket = browser_ticket_backend(&api).unwrap().create(input).unwrap();
let old_coder = seed_live_test_coder_assignment(&api, &ticket.id, "old-coder").await;
// The retained delegation precedes actual Coder work. A new Assign
// during that work would be a different operation and must be refused.
assign_test_orchestrator(&api, &ticket.id);
let old_coder = seed_live_test_coder_assignment(&api, &ticket.id, "old-coder").await;
let old_orchestrator_id = format!("orchestrator-{}", ticket.id);
let Json(started) = scoped_start_workspace_orchestrator(
State(api.clone()),
@@ -48795,6 +49091,9 @@ mod tests {
assigned_by: "test-user".to_string(),
assigned_at: TEST_CREATED_AT.to_string(),
};
// Seed an existing delegation before starting the Worker; both may
// remain current, but a fresh delegation during Worker work is guarded.
assign_test_orchestrator(&api, &ticket_id);
api.store
.set_current_ticket_worker_assignment(
&assignment,
@@ -48804,7 +49103,6 @@ mod tests {
false,
)
.unwrap();
assign_test_orchestrator(&api, &ticket_id);
let path = || ScopedRecordPath {
workspace_id: TEST_WORKSPACE_ID.to_string(),
id: ticket_id.clone(),
@@ -52879,6 +53177,7 @@ mod tests {
worker_id: "legacy-worker".to_string(),
}),
Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: "runtime-source".to_string(),
worker_id: Some("runtime-worker".to_string()),
}),
@@ -52891,6 +53190,7 @@ mod tests {
contract_workdir_removal_source_actor(&context(
None,
Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: "runtime-source".to_string(),
worker_id: Some("runtime-worker".to_string()),
}),
@@ -52903,6 +53203,7 @@ mod tests {
contract_workdir_removal_source_actor(&context(
None,
Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: "runtime-only".to_string(),
worker_id: None,
}),
@@ -52914,6 +53215,7 @@ mod tests {
let error = contract_workdir_removal_source_actor(&context(
None,
Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: "runtime-only".to_string(),
worker_id: None,
}),
@@ -61118,6 +61420,7 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
actor: None,
worker_source: None,
runtime_source: Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: dedicated.worker.runtime_id.clone(),
worker_id: Some(dedicated.worker.worker_id.clone()),
}),
@@ -61141,6 +61444,7 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
actor: None,
worker_source: None,
runtime_source: Some(server_api::ServerRuntimeSource {
authentication_binding_id: "fixture-binding".into(),
runtime_id: forged.worker.runtime_id.clone(),
worker_id: Some(forged.worker.worker_id.clone()),
}),
crates/workspace-server/src/server/tests/drive_api_tests.rs
+2
View File
@@ -721,6 +721,7 @@ async fn drive_verified_upload_revoked_during_reception_cannot_commit_or_replay(
token_id_hash: None,
runtime_id: workers[0].runtime_id.clone(),
worker_id: Some(workers[0].worker_id.clone()),
authentication_binding_id: "fixture-binding".into(),
},
));
let response = app
@@ -867,6 +868,7 @@ async fn drive_read_entrypoints_use_current_grants_and_cannot_fallback_to_browse
context.runtime_source = Some(ServerRuntimeSource {
runtime_id: workers[0].runtime_id.clone(),
worker_id: Some(workers[1].worker_id.clone()),
authentication_binding_id: "fixture-binding".into(),
});
assert_eq!(
drive::root(&api, &context, TEST_WORKSPACE_ID)
crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs
+1
View File
@@ -53,6 +53,7 @@ fn context(worker: &RuntimeWorkerRef) -> server_api::ServerRequestContext {
runtime_source: Some(server_api::ServerRuntimeSource {
runtime_id: worker.runtime_id.clone(),
worker_id: Some(worker.worker_id.clone()),
authentication_binding_id: "fixture-binding".into(),
}),
origin: None,
transport_headers: vec![],
crates/workspace-server/src/server/tests/worker_operations_tests.rs
+4
View File
@@ -632,12 +632,14 @@ async fn request_context_uses_verified_subject_not_headers_or_account_fallback()
assert!(WorkerOperationContext::from_request(&api, &headers_only).is_err());
let mut runtime_only = operation_request_context(Some(server_api::ServerRuntimeSource {
runtime_id: "runtime-a".into(),
authentication_binding_id: "fixture-binding".into(),
worker_id: None,
}));
runtime_only.actor = Some(test_browser_request_actor());
assert!(WorkerOperationContext::from_request(&api, &runtime_only).is_err());
let mut mismatched = operation_request_context(Some(server_api::ServerRuntimeSource {
runtime_id: "runtime-a".into(),
authentication_binding_id: "fixture-binding".into(),
worker_id: Some("controller-a".into()),
}));
mismatched.actor = Some(test_browser_request_actor());
@@ -646,6 +648,7 @@ async fn request_context_uses_verified_subject_not_headers_or_account_fallback()
seed_worker_source_member(&api, "runtime-a", "controller-a");
let context = operation_request_context(Some(server_api::ServerRuntimeSource {
runtime_id: "runtime-a".into(),
authentication_binding_id: "fixture-binding".into(),
worker_id: Some("controller-a".into()),
}));
assert!(
@@ -665,6 +668,7 @@ async fn generic_management_aliases_cannot_bypass_worker_grants_or_target_bound_
let context = || {
operation_request_context(Some(server_api::ServerRuntimeSource {
runtime_id: "offline-runtime".into(),
authentication_binding_id: "fixture-binding".into(),
worker_id: Some("controller".into()),
}))
};
crates/workspace-server/src/server_assignment_tracking_tests.rs
+1501
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/store.rs
+585 -257
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/ticket_tag_upgrade_tests.rs
+15 -5
View File
@@ -265,10 +265,20 @@ fn assert_retained_classification_upgrade(workspace_version: i64, component: i64
drop(conn);
let store = SqliteWorkspaceStore::open(&path).unwrap();
store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,91);
assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,10);
assert_eq!(current_schema_version(conn)?,92);
assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,11);
assert_eq!(fixture_auth_rows(conn),auth_before);
for (table, rows) in preserved_tables.iter().zip(&preserved) { assert_eq!(&stored_fixture_rows(conn,table),rows,"preserve {table}"); }
for (table, rows) in preserved_tables.iter().zip(&preserved) {
let mut migrated = stored_fixture_rows(conn, table);
if *table == "ticket_worker_assignments" {
// Component 11 adds handling columns to the same identity row.
// Compare the ten retained identity/principal/Assign columns,
// and separately prove the historical Worker acquired no tracking.
for row in &mut migrated { row.truncate(10); }
assert_eq!(conn.query_row("SELECT COUNT(*) FROM ticket_worker_assignments WHERE tracking_json IS NOT NULL", [], |row| row.get::<_,i64>(0))?, 0);
}
assert_eq!(&migrated, rows, "preserve {table}");
}
if let Some(rows)=&tags_before {assert_eq!(&classification_rows(conn),rows);}
assert_eq!(conn.query_row("SELECT COUNT(*) FROM ticket_assignment_work_releases WHERE assignment_id='assignment'",[],|row|row.get::<_,i64>(0))?,1);
assert_eq!(conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check",[],|row|row.get::<_,i64>(0))?,0);
@@ -483,14 +493,14 @@ fn workspace90_cases9_add_classification_without_reinterpreting_reasoned_history
let store = SqliteWorkspaceStore::open(&path).unwrap();
store
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 91);
assert_eq!(current_schema_version(conn)?, 92);
assert_eq!(
conn.query_row(
"SELECT MAX(version) FROM ticket_schema_migrations",
[],
|row| row.get::<_, i64>(0)
)?,
10
11
);
assert_eq!(
stored_fixture_rows(conn, "typed_ticket_case_history"),
crates/workspace-server/src/worker_source/runtime_request.rs
+4
View File
@@ -12,6 +12,9 @@ use crate::store::ControlPlaneStore;
pub struct VerifiedRuntimeRequestSource {
pub runtime_id: String,
pub worker_id: Option<String>,
/// The binding actually used for signature verification. Mutation boundaries
/// compare this identity with current trust inside their write transaction.
pub authentication_binding_id: String,
/// Existing proof JTI correlation, retained only as a fixed-length diagnostic fingerprint.
pub token_id_hash: Option<String>,
}
@@ -273,5 +276,6 @@ pub(crate) async fn verify_runtime_request_source_proof_with_clock(
token_id_hash: Some(diagnostic_id_hash(&claims.jti)),
runtime_id: claims.iss,
worker_id: claims.worker_id,
authentication_binding_id: trusted.binding_id,
})
}
crates/workspace-server/src/workspace_deletion.rs
+79 -10
View File
@@ -52,6 +52,8 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
"ticket_current_worker_assignments",
"ticket_worker_assignment_events",
"ticket_worker_assignments",
"typed_ticket_assignment_tracking_history",
"typed_ticket_assignment_tracking_operations",
"typed_ticket_case_history",
"typed_ticket_case_metadata",
"typed_ticket_legacy_workflow_snapshot",
@@ -464,15 +466,21 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
}
let workspace_id = operation.response.workspace_id.clone();
// Case judgments and legacy snapshots are immutable under ordinary
// Ticket operations, including Ticket deletion and FK cascades. Only
// this owner-authorized whole-Workspace purge may discard them.
// Ticket case/assignment judgments and their operation receipts are
// immutable under ordinary domain operations, including Ticket
// deletion and FK cascades. Only this owner-authorized whole-Workspace
// purge may discard them.
// SQLite transactional DDL holds the writer lock: no concurrent writer
// can observe the suspended guards, and any error rolls back both
// data and schema. Restore the exact schema-owned definitions before
// commit; update guards remain installed throughout.
let mut case_delete_guards = Vec::new();
let mut ticket_delete_guards = Vec::new();
for (table, trigger) in [
("ticket_worker_assignments", "typed_ticket_assignment_identity_no_delete"),
("ticket_worker_assignment_events", "typed_ticket_assignment_events_no_delete"),
("ticket_assignment_operations", "typed_ticket_assignment_receipt_no_delete"),
("typed_ticket_assignment_tracking_history", "typed_ticket_assignment_tracking_history_no_delete"),
("typed_ticket_assignment_tracking_operations", "typed_ticket_assignment_tracking_operations_no_delete"),
("typed_ticket_case_history", "typed_ticket_case_history_no_delete"),
("typed_ticket_legacy_workflow_snapshot", "typed_ticket_legacy_workflow_no_delete"),
] {
@@ -482,7 +490,7 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
|row| row.get(0),
)?;
tx.execute_batch(&format!("DROP TRIGGER \"{trigger}\";"))?;
case_delete_guards.push(definition);
ticket_delete_guards.push(definition);
}
for table in WORKSPACE_DELETION_PURGE_TABLES {
tx.execute(
@@ -499,7 +507,7 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
"Workspace disappeared before deletion finalized".to_string(),
));
}
for definition in case_delete_guards {
for definition in ticket_delete_guards {
tx.execute_batch(&definition)?;
}
let now = Utc::now().to_rfc3339();
@@ -978,6 +986,57 @@ mod tests {
.create_with_snapshot(ticket::NewTicket::new("Case evidence"))
.unwrap()
.1;
let assignment_id = format!("delegation-{workspace}");
store
.set_current_ticket_role_assignment(
&crate::store::TicketRoleAssignmentRecord {
workspace_id: workspace.into(),
ticket_id: initial.meta.id.clone(),
assignment_id: assignment_id.clone(),
role: crate::store::TicketAssignmentRole::Orchestrator,
principal: crate::store::TicketAssignmentPrincipal::WorkspaceAgent {
agent_key: "workspace-orchestrator".into(),
},
assigned_by: "owner".into(),
assigned_at: "2026-01-01T00:00:00Z".into(),
},
None,
"assigned",
"assign",
false,
)
.unwrap();
// Seed domain handling evidence under a transaction. This fixture
// exercises owner purge, not singleton authentication or delivery.
store
.with_transaction(|tx| {
ticket::sqlite_update_ticket_assignment_tracking_in_transaction(
tx,
workspace,
&initial.meta.id,
&ticket::TicketAssignmentTrackingUpdate {
assignment_id: assignment_id.clone(),
expected_last_event_id: "assigned".into(),
operation_key: "reported".into(),
tracking: ticket::TicketAssignmentTracking {
status: ticket::TicketAssignmentTrackingStatus::Done,
waiting_reason: None,
progress: None,
next_action: None,
result: Some("Returned result".into()),
},
},
&ticket::TicketAssignmentTrackingSourceContext {
actor: "orchestrator".into(),
runtime_id: "fixture-runtime".into(),
worker_id: "fixture-worker".into(),
agent_key: "workspace-orchestrator".into(),
authentication_binding_id: "fixture-binding".into(),
},
)?;
Ok(())
})
.unwrap();
backend
.update_case(
&initial.meta.id,
@@ -1022,7 +1081,7 @@ mod tests {
}
let assert_immutable = |workspace: &str| {
store.with_conn(|conn| {
for table in ["typed_ticket_case_history", "typed_ticket_legacy_workflow_snapshot"] {
for table in ["typed_ticket_case_history", "typed_ticket_legacy_workflow_snapshot", "ticket_worker_assignment_events", "typed_ticket_assignment_tracking_history", "typed_ticket_assignment_tracking_operations", "ticket_assignment_operations"] {
assert!(conn.execute(&format!("DELETE FROM {table} WHERE workspace_id=?1"), [workspace]).is_err());
assert!(conn.execute(&format!("UPDATE {table} SET ticket_id=ticket_id WHERE workspace_id=?1"), [workspace]).is_err());
}
@@ -1052,8 +1111,13 @@ mod tests {
"typed_ticket_case_history",
"typed_ticket_case_metadata",
"typed_ticket_legacy_workflow_snapshot",
"ticket_worker_assignments",
"ticket_worker_assignment_events",
"ticket_assignment_operations",
"typed_ticket_assignment_tracking_history",
"typed_ticket_assignment_tracking_operations",
];
let other_before: Vec<String> = store
let other_before: Vec<Option<String>> = store
.with_conn(|conn| {
tables
.iter()
@@ -1095,7 +1159,12 @@ mod tests {
[],
|row| row.get(0),
)?;
assert_eq!(count, 1, "rollback must restore {table}");
let expected = if table == "typed_ticket_assignment_tracking_history" {
2
} else {
1
};
assert_eq!(count, expected, "rollback must restore {table}");
}
conn.execute_batch("DROP TRIGGER reject_workspace_purge;")?;
Ok(())
@@ -1132,7 +1201,7 @@ mod tests {
|row| row.get(0),
)?;
assert_eq!(count, 0, "purge must remove {table}");
let after: String = conn.query_row(
let after: Option<String> = conn.query_row(
&format!("SELECT * FROM {table} WHERE workspace_id='workspace-b'"),
[],
|row| row.get(row.as_ref().column_count() - 1),