feat: integrate committed Ticket event services

crates/workspace-server/Cargo.toml

@@ -66,6 +66,7 @@ webauthn-rs = { workspace = true, features = ["conditional-ui"] }
 webauthn-rs-proto = "0.5.5"
 
 [dev-dependencies]
+feature-storage = { workspace = true, features = ["test-support"] }
 native-tls = "0.2"
 tokio-native-tls = "0.3"
 wip-protocol.workspace = true

crates/workspace-server/src/assignment_tracking_tests.rs

@@ -525,7 +525,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip
                     (
                         111,
                         "Typed Plugin Ticket extension data and atomic mutation receipts".into()
-                    )
+                    ),
+                    (112, "Ticket Service outbox and fenced consumers".into())
                 ]
             );
             for (table, rows) in retained.iter().zip(original.iter()) {

crates/workspace-server/src/lib.rs

@@ -20,6 +20,7 @@ pub mod plugin_host;
 pub mod plugin_https;
 pub mod plugin_plan;
 pub mod plugin_receipt;
+pub mod plugin_service;
 pub mod plugin_settings;
 pub mod profile_settings;
 pub mod prompt_settings;

crates/workspace-server/src/plugin_catalog.rs

@@ -258,6 +258,7 @@ fn artifact_metadata(
                 external_write: t.external_write,
             })
             .collect(),
+        services: decode(&encode(&m.services)?)?,
         connections: decode(&encode(&m.connections)?)?,
         requested_grants: decode(&encode(&m.requested_grants)?)?,
         data_collections: decode(&encode(&m.data_collections)?)?,

crates/workspace-server/src/plugin_catalog/migration_tests.rs

@@ -30,7 +30,7 @@ fn plugin_catalog_migration_matches_fresh_schema_from_oldest_retained_and_105()
             // Plugin catalog remains migration 106 even when startup applies later steps.
             assert_eq!(after.iter().filter(|(version, _)| *version == 106).cloned().collect::<Vec<_>>(),
                 vec![(106, crate::plugin_catalog::MIGRATION_NAME.into())]);
-            assert_eq!(after.last().unwrap(), &(111, "Typed Plugin Ticket extension data and atomic mutation receipts".into()));
+            assert_eq!(after.last().unwrap(), &(112, "Ticket Service outbox and fenced consumers".into()));
             assert_eq!(current_schema_version(c)?, LATEST_SCHEMA_VERSION);
             assert_eq!(c.query_row("SELECT display_name FROM workspaces WHERE workspace_id='plugin-workspace'",[],|r|r.get::<_,String>(0))?,"Plugins");
             assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
@@ -81,6 +81,7 @@ fn plugin_catalog_migration_marker_failure_rolls_back_then_retries_without_chang
                 111,
                 "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
             ));
+            expected.push((112, "Ticket Service outbox and fenced consumers".into()));
             assert_eq!(workspace_schema_migration_history(c)?, expected);
             crate::plugin_catalog::verify_with_data(c)?;
             Ok(())

crates/workspace-server/src/plugin_data/tests.rs

@@ -45,6 +45,7 @@ fn plan(workspace: &str, installation: &str, id: &str) -> StoredPluginPlan {
         PluginPinProjection {
             config: serde_json::json!({}),
             surfaces: vec![PluginSurface::Tool],
+            services: vec![],
             tools: vec!["echo".into()],
             grants: vec![],
             bindings: vec![],

crates/workspace-server/src/plugin_host.rs

@@ -24,6 +24,7 @@ pub struct AdmittedPluginOperation {
     pub pin: WorkerPluginPin,
     pub binding: Option<PluginConnectionBinding>,
     pub data_collections: Vec<PluginDataSchemaMetadata>,
+    pub services: Vec<PluginServiceMetadata>,
 }
 
 /// `execution` is derived by the authenticated Host boundary from current Worker
@@ -38,6 +39,47 @@ pub fn authorize(
     installation: &str,
     tools: &[String],
     operation: &PluginHostOperation,
+) -> Result<AdmittedPluginOperation> {
+    authorize_selected(
+        tx,
+        workspace,
+        execution,
+        plan_id,
+        installation,
+        tools,
+        None,
+        operation,
+    )
+}
+pub fn authorize_service(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    execution: &HostExecutionIdentity,
+    plan_id: &str,
+    installation: &str,
+    service: &str,
+    operation: &PluginHostOperation,
+) -> Result<AdmittedPluginOperation> {
+    authorize_selected(
+        tx,
+        workspace,
+        execution,
+        plan_id,
+        installation,
+        &[],
+        Some(service),
+        operation,
+    )
+}
+fn authorize_selected(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    execution: &HostExecutionIdentity,
+    plan_id: &str,
+    installation: &str,
+    tools: &[String],
+    service: Option<&str>,
+    operation: &PluginHostOperation,
 ) -> Result<AdmittedPluginOperation> {
     let plan = plugin_plan_load(tx, workspace, plan_id)?;
     if plan.plan.execution() != execution || plan.pin_state == PlanPinState::Released {
@@ -50,20 +92,31 @@ pub fn authorize(
         .find(|pin| pin.installation_id() == installation)
         .cloned()
         .ok_or_else(|| failure(PluginHostErrorCode::Denied))?;
-    if tools.is_empty()
-        || tools.len() > 256
-        || tools
-            .iter()
-            .any(|tool| tool.is_empty() || tool.len() > 512 || tool.chars().any(char::is_control))
-        || tools
-            .iter()
-            .collect::<std::collections::BTreeSet<_>>()
-            .len()
-            != tools.len()
+    if service.is_none()
+        && (tools.is_empty()
+            || tools.len() > 256
+            || tools.iter().any(|tool| {
+                tool.is_empty() || tool.len() > 512 || tool.chars().any(char::is_control)
+            })
+            || tools
+                .iter()
+                .collect::<std::collections::BTreeSet<_>>()
+                .len()
+                != tools.len())
     {
         return Err(failure(PluginHostErrorCode::InvalidInput));
     }
-    if !pin.projection().surfaces.contains(&PluginSurface::Tool)
+    if let Some(service) = service {
+        if service.is_empty()
+            || service.len() > 512
+            || service.chars().any(char::is_control)
+            || !pin.projection().surfaces.contains(&PluginSurface::Service)
+            || !pin.projection().services.iter().any(|s| s == service)
+            || !pin.projection().grants.contains(&PluginGrant::TicketEvents)
+        {
+            return Err(failure(PluginHostErrorCode::Denied));
+        }
+    } else if !pin.projection().surfaces.contains(&PluginSurface::Tool)
         || tools
             .iter()
             .any(|tool| !pin.projection().tools.contains(tool))
@@ -137,6 +190,7 @@ pub fn authorize(
         pin,
         binding,
         data_collections: artifact.data_collections,
+        services: artifact.services,
     })
 }
 

crates/workspace-server/src/plugin_host_migration_tests.rs

@@ -12,7 +12,7 @@ fn plugin_receipt_tail_migrates_true_predecessor_reopens_and_matches_fresh_schem
     assert!(!table_exists(&conn, "plugin_invocations").unwrap());
     let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
     assert_eq!(plan.current_schema_version, 108);
-    assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109, 110, 111]);
+    assert_eq!(plan.migrations.iter().map(|step| step.version).collect::<Vec<_>>(), [109, 110, 111, 112]);
     assert_eq!(current_schema_version(&conn).unwrap(), 108);
     drop(conn);
     let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();

crates/workspace-server/src/plugin_plan/migration_tests.rs

@@ -39,6 +39,7 @@ fn plugin_plan_migration_marker_failure_preserves_catalog_then_retries_and_match
                 111,
                 "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
             ));
+            expected.push((112, "Ticket Service outbox and fenced consumers".into()));
             assert_eq!(workspace_schema_migration_history(conn)?, expected);
             subject_assignment_migration::verify(conn)?;
             crate::plugin_plan::verify_plan_schema(conn)?;

crates/workspace-server/src/plugin_plan/tests.rs

@@ -130,6 +130,7 @@ fn projection() -> PluginPinProjection {
     PluginPinProjection {
         config: serde_json::json!({}),
         surfaces: vec![PluginSurface::Tool],
+        services: vec![],
         tools: vec!["example_echo".into()],
         grants: vec![],
         bindings: vec![],

crates/workspace-server/src/plugin_receipt.rs

@@ -46,6 +46,8 @@ pub struct InvocationRecord {
     pub expires_at_ms: i64,
     pub cancelled: bool,
     pub expired: bool,
+    /// Ephemeral proof supplied by the authenticated Service wrapper, never persisted.
+    pub service_lease: Option<server_api::PluginServiceLease>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -222,6 +224,7 @@ pub fn lookup(conn: &Connection, workspace: &str, invocation_id: &str) -> Result
         expires_at_ms,
         cancelled,
         expired,
+        service_lease: None,
     })
 }
 
@@ -231,7 +234,7 @@ pub(crate) fn verify_record(
     conn: &Connection,
     supplied: &InvocationRecord,
 ) -> Result<InvocationRecord> {
-    let saved = lookup(conn, &supplied.workspace_id, &supplied.invocation_id)?;
+    let mut saved = lookup(conn, &supplied.workspace_id, &supplied.invocation_id)?;
     if supplied.plan_id != saved.plan_id
         || supplied.installation_id != saved.installation_id
         || supplied.execution != saved.execution
@@ -240,6 +243,13 @@ pub(crate) fn verify_record(
     {
         return conflict();
     }
+    crate::plugin_service::fence_invocation(
+        conn,
+        supplied,
+        supplied.service_lease.as_ref(),
+        chrono::Utc::now().timestamp_millis(),
+    )?;
+    saved.service_lease = supplied.service_lease.clone();
     Ok(saved)
 }
 fn is_expired(conn: &Connection, record: &InvocationRecord) -> Result<bool> {

crates/workspace-server/src/plugin_receipt/tests.rs

@@ -138,6 +138,7 @@ fn make_plan(
             error_codes: vec![],
             external_write: false,
         }],
+        services: vec![],
         connections: vec![],
         requested_grants: vec![],
         data_collections: vec![],
@@ -158,6 +159,7 @@ fn make_plan(
         PluginPinProjection {
             config: serde_json::json!({}),
             surfaces: vec![PluginSurface::Tool],
+            services: vec![],
             tools: vec!["example_echo".into()],
             grants: vec![],
             bindings: vec![],

crates/workspace-server/src/plugin_service.rs

@@ -0,0 +1,518 @@
+//! Durable domain delivery state, not a task registry. The Runtime owns the driver
+//! task; this owner serializes admission, lease proofs, receipts and cursor commits
+//! in the catalog database. No guest or network I/O runs inside these transactions.
+use crate::{Result, plugin_host::failure, plugin_plan::StoredPluginPlan, plugin_receipt};
+use rusqlite::{Connection, OptionalExtension, Transaction, params};
+use server_api::*;
+use sha2::{Digest, Sha256};
+
+pub const SCHEMA_SQL: &str = include_str!("plugin_service_schema.sql");
+pub const LEASE_MS: i64 = 60_000;
+pub const MAX_QUEUE: u32 = 256;
+pub const MAX_ATTEMPTS: u32 = 5;
+pub const MAX_RETRY_MS: u32 = 300_000;
+pub const RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1000;
+
+#[derive(Clone, Debug)]
+pub struct Consumer {
+    pub consumer_id: String,
+    pub workspace: String,
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub installation: String,
+    pub plan_id: String,
+    pub service: String,
+    pub subscriptions: Vec<PluginTicketEventType>,
+    pub cursor: i64,
+    pub drain_through: i64,
+    pub pause: Option<PluginServicePauseReason>,
+    pub current_sequence: Option<i64>,
+    pub current_event_id: Option<String>,
+    pub invocation_id: Option<String>,
+    pub attempt: u32,
+    pub next_attempt_ms: i64,
+    pub lease_token: Option<String>,
+    pub lease_until_ms: i64,
+}
+fn decode<T: serde::de::DeserializeOwned>(json: &str) -> Result<T> {
+    serde_json::from_str(json).map_err(|_| failure(PluginHostErrorCode::Unavailable))
+}
+fn encode<T: serde::Serialize>(value: &T) -> Result<String> {
+    serde_json::to_string(value).map_err(|_| failure(PluginHostErrorCode::InvalidInput))
+}
+fn valid(s: &str, max: usize) -> bool {
+    !s.is_empty() && s.len() <= max && !s.chars().any(char::is_control)
+}
+
+pub fn load(conn: &Connection, workspace: &str, id: &str) -> Result<Consumer> {
+    if !valid(id, 512) {
+        return Err(failure(PluginHostErrorCode::InvalidInput));
+    }
+    let row = conn.query_row("SELECT runtime_id,worker_id,installation_id,plan_id,service_name,subscriptions_json,cursor,pause_reason,current_sequence,current_event_id,invocation_id,attempt,next_attempt_ms,lease_token,lease_until_ms,drain_through FROM plugin_service_consumers WHERE workspace_id=?1 AND consumer_id=?2", params![workspace,id], |r| Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,String>(5)?,r.get::<_,i64>(6)?,r.get::<_,Option<String>>(7)?,r.get::<_,Option<i64>>(8)?,r.get::<_,Option<String>>(9)?,r.get::<_,Option<String>>(10)?,r.get::<_,u32>(11)?,r.get::<_,i64>(12)?,r.get::<_,Option<String>>(13)?,r.get::<_,i64>(14)?,r.get::<_,i64>(15)?))).optional()?.ok_or_else(||failure(PluginHostErrorCode::NotFound))?;
+    Ok(Consumer {
+        consumer_id: id.into(),
+        workspace: workspace.into(),
+        runtime_id: row.0,
+        worker_id: row.1,
+        installation: row.2,
+        plan_id: row.3,
+        service: row.4,
+        subscriptions: decode(&row.5)?,
+        cursor: row.6,
+        drain_through: row.15,
+        pause: row.7.map(|v| decode(&format!("\"{v}\""))).transpose()?,
+        current_sequence: row.8,
+        current_event_id: row.9,
+        invocation_id: row.10,
+        attempt: row.11,
+        next_attempt_ms: row.12,
+        lease_token: row.13,
+        lease_until_ms: row.14,
+    })
+}
+fn floor(conn: &Connection, workspace: &str) -> Result<i64> {
+    Ok(conn
+        .query_row(
+            "SELECT removed_through FROM plugin_service_retention WHERE workspace_id=?1",
+            [workspace],
+            |r| r.get(0),
+        )
+        .optional()?
+        .unwrap_or(0))
+}
+fn tail(conn: &Connection, workspace: &str) -> Result<i64> {
+    Ok(conn.query_row("SELECT COALESCE(MAX(sequence),0) FROM typed_ticket_service_outbox WHERE workspace_id=?1",[workspace],|r|r.get::<_,i64>(0))?.max(floor(conn,workspace)?))
+}
+
+pub fn acquire(
+    tx: &Transaction<'_>,
+    plan: &StoredPluginPlan,
+    installation: &str,
+    service: &str,
+    subscriptions: &[PluginTicketEventType],
+    now: i64,
+) -> Result<PluginServiceLease> {
+    if !valid(service, 512)
+        || subscriptions.is_empty()
+        || subscriptions.len() > 2
+        || (subscriptions.len() == 2 && subscriptions[0] == subscriptions[1])
+    {
+        return Err(failure(PluginHostErrorCode::InvalidInput));
+    }
+    let workspace = plan.plan.workspace_id();
+    let execution = plan.plan.execution();
+    let existing:Option<String>=tx.query_row("SELECT consumer_id FROM plugin_service_consumers WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND installation_id=?4 AND plan_id=?5 AND service_name=?6",params![workspace,execution.runtime_id,execution.worker_id,installation,plan.plan_id,service],|r|r.get(0)).optional()?;
+    let id = if let Some(id) = existing {
+        let saved = load(tx, workspace, &id)?;
+        if saved.subscriptions != subscriptions {
+            return Err(failure(PluginHostErrorCode::Conflict));
+        }
+        if saved.lease_token.is_some() && saved.lease_until_ms > now {
+            return Err(failure(PluginHostErrorCode::Conflict));
+        }
+        id
+    } else {
+        let count: u32 = tx.query_row(
+            "SELECT count(*) FROM plugin_service_consumers WHERE workspace_id=?1",
+            [workspace],
+            |r| r.get(0),
+        )?;
+        if count >= 1024 {
+            return Err(failure(PluginHostErrorCode::QuotaExceeded));
+        }
+        let id = uuid::Uuid::now_v7().to_string();
+        tx.execute("INSERT INTO plugin_service_consumers(workspace_id,consumer_id,runtime_id,worker_id,installation_id,plan_id,service_name,subscriptions_json,cursor) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9)",params![workspace,id,execution.runtime_id,execution.worker_id,installation,plan.plan_id,service,encode(&subscriptions)?,tail(tx,workspace)?])?;
+        id
+    };
+    let token = uuid::Uuid::new_v4().to_string();
+    tx.execute("UPDATE plugin_service_consumers SET lease_token=?3,lease_until_ms=?4 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,id,token,now+LEASE_MS])?;
+    Ok(PluginServiceLease {
+        consumer_id: id,
+        lease_token: token,
+    })
+}
+
+pub fn fence(
+    conn: &Connection,
+    workspace: &str,
+    lease: &PluginServiceLease,
+    now: i64,
+) -> Result<Consumer> {
+    let c = load(conn, workspace, &lease.consumer_id)?;
+    if !valid(&lease.lease_token, 512)
+        || c.lease_token.as_deref() != Some(lease.lease_token.as_str())
+        || c.lease_until_ms <= now
+    {
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    Ok(c)
+}
+
+/// Called from receipt verification in the same transaction as every effect/data
+/// mutation. A stable logical invocation never suffices as an old driver's proof.
+pub fn fence_invocation(
+    conn: &Connection,
+    record: &plugin_receipt::InvocationRecord,
+    lease: Option<&PluginServiceLease>,
+    now: i64,
+) -> Result<()> {
+    if let PluginConsumer::Consumer {
+        consumer_id,
+        event_id,
+        export_name,
+    } = &record.consumer
+    {
+        let lease = lease.ok_or_else(|| failure(PluginHostErrorCode::Denied))?;
+        let c = fence(conn, &record.workspace_id, lease, now)?;
+        if &c.consumer_id != consumer_id
+            || c.current_event_id.as_ref() != Some(event_id)
+            || &c.service != export_name
+            || c.invocation_id.as_deref() != Some(record.invocation_id.as_str())
+            || c.pause.is_some()
+            || c.plan_id != record.plan_id
+            || c.installation != record.installation_id
+        {
+            return Err(failure(PluginHostErrorCode::Denied));
+        }
+    } else if lease.is_some() {
+        return Err(failure(PluginHostErrorCode::Denied));
+    }
+    Ok(())
+}
+fn queued(conn: &Connection, c: &Consumer) -> Result<u32> {
+    let changed = c.subscriptions.contains(&PluginTicketEventType::Changed);
+    let deleted = c.subscriptions.contains(&PluginTicketEventType::Deleted);
+    Ok(conn.query_row("SELECT count(*) FROM (SELECT 1 FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence>?2 AND ((event_type='changed' AND ?3) OR (event_type='deleted' AND ?4)) ORDER BY sequence LIMIT 257)",params![c.workspace,c.cursor,changed,deleted],|r|r.get(0))?)
+}
+pub fn diagnostics(conn: &Connection, c: &Consumer, now: i64) -> Result<PluginServiceDiagnostics> {
+    Ok(PluginServiceDiagnostics {
+        consumer_id: c.consumer_id.clone(),
+        pause_reason: c.pause,
+        queued: queued(conn, c)?,
+        attempt: c.attempt,
+        current_event_id: match (&c.current_event_id, c.pause) {
+            (Some(id), _) => Some(id.clone()),
+            (None, Some(PluginServicePauseReason::QueueFull)) => conn.query_row(
+                "SELECT event_id FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence>?2 AND ((event_type='changed' AND ?3) OR (event_type='deleted' AND ?4)) ORDER BY sequence LIMIT 1",
+                params![c.workspace, c.cursor, c.subscriptions.contains(&PluginTicketEventType::Changed), c.subscriptions.contains(&PluginTicketEventType::Deleted)],
+                |row| row.get(0),
+            ).optional()?,
+            _ => None,
+        },
+        lease_active: c.lease_token.is_some() && c.lease_until_ms > now,
+    })
+}
+pub fn pause(tx: &Transaction<'_>, c: &Consumer, reason: PluginServicePauseReason) -> Result<()> {
+    let reason = encode(&reason)?;
+    tx.execute("UPDATE plugin_service_consumers SET pause_reason=?3 WHERE workspace_id=?1 AND consumer_id=?2",params![c.workspace,c.consumer_id,reason.trim_matches('"')])?;
+    Ok(())
+}
+fn unknown_effect(conn: &Connection, c: &Consumer, now: i64) -> Result<bool> {
+    let Some(id) = &c.invocation_id else {
+        return Ok(false);
+    };
+    Ok(conn.query_row("SELECT EXISTS(SELECT 1 FROM plugin_effect_receipts WHERE workspace_id=?1 AND invocation_id=?2 AND (state IN ('pending','expired') OR json_extract(error_json,'$.effect')='indeterminate')) OR EXISTS(SELECT 1 FROM plugin_invocations WHERE workspace_id=?1 AND invocation_id=?2 AND (expired=1 OR expires_at_ms<=?3))",params![c.workspace,id,now],|r|r.get(0))?)
+}
+
+pub fn poll(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    lease: &PluginServiceLease,
+    plan: &StoredPluginPlan,
+    now: i64,
+) -> Result<PluginServiceResponse> {
+    let c = fence(tx, workspace, lease, now)?;
+    // Retention is a Host operation in the same owner transaction, not a guest
+    // callback. Persisting the floor makes expiry visible to every consumer.
+    prune_outbox(tx, workspace, tail(tx, workspace)?, now)?;
+    if c.pause.is_some() {
+        return Ok(PluginServiceResponse::Diagnostics {
+            diagnostics: diagnostics(tx, &c, now)?,
+        });
+    }
+    let reason = if c.cursor < floor(tx, workspace)? {
+        Some(PluginServicePauseReason::RetentionGap)
+    } else if queued(tx, &c)? >= MAX_QUEUE && c.cursor >= c.drain_through {
+        Some(PluginServicePauseReason::QueueFull)
+    } else if unknown_effect(tx, &c, now)? {
+        Some(PluginServicePauseReason::Indeterminate)
+    } else {
+        None
+    };
+    if let Some(reason) = reason {
+        pause(tx, &c, reason)?;
+        return Ok(PluginServiceResponse::Diagnostics {
+            diagnostics: diagnostics(tx, &load(tx, workspace, &c.consumer_id)?, now)?,
+        });
+    }
+    tx.execute("UPDATE plugin_service_consumers SET lease_until_ms=?3 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,c.consumer_id,now+LEASE_MS])?;
+    if c.next_attempt_ms > now {
+        return Ok(PluginServiceResponse::Idle);
+    }
+    if c.current_event_id.is_some() && c.attempt >= MAX_ATTEMPTS {
+        pause(tx, &c, PluginServicePauseReason::AttemptsExhausted)?;
+        return Ok(PluginServiceResponse::Diagnostics {
+            diagnostics: diagnostics(tx, &load(tx, workspace, &c.consumer_id)?, now)?,
+        });
+    }
+    let event=tx.query_row("SELECT sequence,event_id,event_type,occurred_at,ticket_id,domain_event_ref,content_digest,correlation_id,causation_id FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence>?2 AND ((event_type='changed' AND ?3) OR (event_type='deleted' AND ?4)) ORDER BY sequence LIMIT 1",params![workspace,c.cursor,c.subscriptions.contains(&PluginTicketEventType::Changed),c.subscriptions.contains(&PluginTicketEventType::Deleted)],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,String>(5)?,r.get::<_,String>(6)?,r.get::<_,String>(7)?,r.get::<_,Option<String>>(8)?))).optional()?;
+    let Some((
+        sequence,
+        event_id,
+        kind,
+        occurred_at,
+        ticket_id,
+        domain_event_ref,
+        content_digest,
+        correlation_id,
+        causation_id,
+    )) = event
+    else {
+        return Ok(PluginServiceResponse::Idle);
+    };
+    if c.current_event_id
+        .as_ref()
+        .is_some_and(|id| id != &event_id)
+    {
+        pause(tx, &c, PluginServicePauseReason::RetentionGap)?;
+        return Ok(PluginServiceResponse::Diagnostics {
+            diagnostics: diagnostics(tx, &load(tx, workspace, &c.consumer_id)?, now)?,
+        });
+    }
+    let record = plugin_receipt::begin(
+        tx,
+        plan,
+        &c.installation,
+        &PluginConsumer::Consumer {
+            consumer_id: c.consumer_id.clone(),
+            event_id: event_id.clone(),
+            export_name: c.service.clone(),
+        },
+        now,
+        RETENTION_MS,
+    )?;
+    let attempt = c.attempt + 1;
+    tx.execute("UPDATE plugin_service_consumers SET current_sequence=?3,current_event_id=?4,invocation_id=?5,attempt=?6,next_attempt_ms=?7 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,c.consumer_id,sequence,event_id,record.invocation_id,attempt,now+30_000])?;
+    let envelope = PluginTicketEventEnvelope {
+        event_id,
+        event_type: decode(&format!("\"{kind}\""))?,
+        version: 1,
+        occurred_at,
+        ticket_id,
+        domain_event_ref,
+        content_digest,
+        correlation_id,
+        causation_id,
+        attempt,
+    };
+    if encode(&envelope)?.len() > 64 * 1024 {
+        return Err(failure(PluginHostErrorCode::QuotaExceeded));
+    }
+    Ok(PluginServiceResponse::Delivery {
+        envelope,
+        invocation_id: record.invocation_id,
+        expires_at_ms: record.expires_at_ms,
+    })
+}
+
+pub fn complete(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    lease: &PluginServiceLease,
+    event: &str,
+    result: &PluginServiceEventResult,
+    now: i64,
+) -> Result<()> {
+    let c = fence(tx, workspace, lease, now)?;
+    if c.current_event_id.as_deref() != Some(event) {
+        let ack:Option<String>=tx.query_row("SELECT last_ack_event_id FROM plugin_service_consumers WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,c.consumer_id],|r|r.get(0))?;
+        if ack.as_deref() == Some(event) && matches!(result, PluginServiceEventResult::Ack) {
+            return Ok(());
+        }
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    if c.pause.is_some() {
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    if unknown_effect(tx, &c, now)? || matches!(result, PluginServiceEventResult::Indeterminate) {
+        return pause(tx, &c, PluginServicePauseReason::Indeterminate);
+    }
+    match result {
+        PluginServiceEventResult::Ack => {
+            tx.execute("UPDATE plugin_service_consumers SET cursor=current_sequence,last_ack_event_id=current_event_id,current_sequence=NULL,current_event_id=NULL,invocation_id=NULL,attempt=0,next_attempt_ms=0 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,c.consumer_id])?;
+        }
+        PluginServiceEventResult::RetryAfter { delay_ms } => {
+            if *delay_ms > MAX_RETRY_MS {
+                return Err(failure(PluginHostErrorCode::InvalidInput));
+            }
+            if c.attempt >= MAX_ATTEMPTS {
+                pause(tx, &c, PluginServicePauseReason::AttemptsExhausted)?;
+            } else {
+                tx.execute("UPDATE plugin_service_consumers SET next_attempt_ms=?3 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,c.consumer_id,now+i64::from(*delay_ms)])?;
+            }
+        }
+        PluginServiceEventResult::Reject => pause(tx, &c, PluginServicePauseReason::Rejected)?,
+        PluginServiceEventResult::Indeterminate => unreachable!(),
+    }
+    Ok(())
+}
+/// Cleanup never authorizes work, so an expired matching token may be released.
+/// The API authenticates the consumer's immutable Worker identity. Exact released
+/// proofs reconcile lost responses without touching any successor lease.
+pub fn release(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    lease: &PluginServiceLease,
+    now: i64,
+) -> Result<()> {
+    if !valid(&lease.consumer_id, 512) || !valid(&lease.lease_token, 512) {
+        return Err(failure(PluginHostErrorCode::InvalidInput));
+    }
+    let released: bool = tx.query_row(
+        "SELECT EXISTS(SELECT 1 FROM plugin_service_released_proofs WHERE workspace_id=?1 AND consumer_id=?2 AND lease_token=?3)",
+        params![workspace, lease.consumer_id, lease.lease_token],
+        |r| r.get(0),
+    )?;
+    if released {
+        return Ok(());
+    }
+    // NULL alone is not proof of this release. Only the exact current token may
+    // be cleared, even if it has expired; an unrecorded stale token must fail.
+    let cleared = tx.execute(
+        "UPDATE plugin_service_consumers SET lease_token=NULL,lease_until_ms=0 WHERE workspace_id=?1 AND consumer_id=?2 AND lease_token=?3",
+        params![workspace, lease.consumer_id, lease.lease_token],
+    )?;
+    if cleared != 1 {
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    tx.execute(
+        "INSERT INTO plugin_service_released_proofs(workspace_id,consumer_id,lease_token,released_at_ms) VALUES(?1,?2,?3,?4)",
+        params![workspace, lease.consumer_id, lease.lease_token, now],
+    )?;
+    Ok(())
+}
+
+/// Caller supplies authenticated Workspace management authority. Audit and cursor
+/// change commit atomically; unknown external effects cannot be resumed blindly.
+pub fn recover(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    id: &str,
+    expected: PluginServicePauseReason,
+    action: &PluginServiceRecoveryAction,
+    key: &str,
+    actor: &str,
+    reason: &str,
+    now: i64,
+) -> Result<()> {
+    if !valid(key, 128) || !valid(actor, 512) || !valid(reason, 1024) {
+        return Err(failure(PluginHostErrorCode::InvalidInput));
+    }
+    let hex: String = Sha256::digest(encode(&(id, expected, action, actor, reason))?)
+        .iter()
+        .map(|b| format!("{b:02x}"))
+        .collect();
+    let fp = format!("sha256:{hex}");
+    let saved:Option<String>=tx.query_row("SELECT fingerprint FROM plugin_service_recovery_audit WHERE workspace_id=?1 AND operation_key=?2",params![workspace,key],|r|r.get(0)).optional()?;
+    if let Some(saved) = saved {
+        return if saved == fp {
+            Ok(())
+        } else {
+            Err(failure(PluginHostErrorCode::Conflict))
+        };
+    }
+    let c = load(tx, workspace, id)?;
+    if c.pause != Some(expected) || (c.lease_token.is_some() && c.lease_until_ms > now) {
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    let cursor = match action {
+        PluginServiceRecoveryAction::Resume => {
+            if matches!(
+                expected,
+                PluginServicePauseReason::Indeterminate | PluginServicePauseReason::RetentionGap
+            ) || unknown_effect(tx, &c, now)?
+                || c.cursor < floor(tx, workspace)?
+                || (expected != PluginServicePauseReason::QueueFull && queued(tx, &c)? >= MAX_QUEUE)
+            {
+                return Err(failure(PluginHostErrorCode::Conflict));
+            }
+            c.cursor
+        }
+        PluginServiceRecoveryAction::Skip { event_id } => {
+            if c.current_event_id.as_ref() == Some(event_id) {
+                c.current_sequence
+                    .ok_or_else(|| failure(PluginHostErrorCode::Conflict))?
+            } else if c.current_event_id.is_none()
+                && expected == PluginServicePauseReason::QueueFull
+            {
+                // Queue admission may stop before an invocation exists. Only the
+                // exact observed queue head can be abandoned, never an arbitrary
+                // later event or a guessed missing retention identity.
+                let head: Option<(i64, String)> = tx.query_row(
+                    "SELECT sequence,event_id FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence>?2 AND ((event_type='changed' AND ?3) OR (event_type='deleted' AND ?4)) ORDER BY sequence LIMIT 1",
+                    params![workspace, c.cursor, c.subscriptions.contains(&PluginTicketEventType::Changed), c.subscriptions.contains(&PluginTicketEventType::Deleted)],
+                    |row| Ok((row.get(0)?, row.get(1)?)),
+                ).optional()?;
+                let Some((sequence, id)) = head else {
+                    return Err(failure(PluginHostErrorCode::Conflict));
+                };
+                if &id != event_id {
+                    return Err(failure(PluginHostErrorCode::Conflict));
+                }
+                sequence
+            } else {
+                return Err(failure(PluginHostErrorCode::Conflict));
+            }
+        }
+        PluginServiceRecoveryAction::AcceptGap => {
+            if expected != PluginServicePauseReason::RetentionGap {
+                return Err(failure(PluginHostErrorCode::Conflict));
+            }
+            floor(tx, workspace)?
+        }
+    };
+    tx.execute("INSERT INTO plugin_service_recovery_audit(workspace_id,operation_key,consumer_id,fingerprint,actor,reason,action_json,at_ms) VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",params![workspace,key,id,fp,actor,reason,encode(action)?,now])?;
+    // Resume preserves the blocked event identity/receipts; skip/gap explicitly
+    // abandons it. Attempts reset only under this audited management decision.
+    if matches!(action, PluginServiceRecoveryAction::Resume) {
+        let drain = if expected == PluginServicePauseReason::QueueFull {
+            tail(tx, workspace)?
+        } else {
+            c.drain_through
+        };
+        tx.execute("UPDATE plugin_service_consumers SET pause_reason=NULL,attempt=0,next_attempt_ms=0,drain_through=?3 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,id,drain])?;
+    } else {
+        tx.execute("UPDATE plugin_service_consumers SET cursor=?3,pause_reason=NULL,current_sequence=NULL,current_event_id=NULL,invocation_id=NULL,attempt=0,next_attempt_ms=0 WHERE workspace_id=?1 AND consumer_id=?2",params![workspace,id,cursor])?;
+    }
+    Ok(())
+}
+
+/// Host retention only: never remove an event before the advertised replay window.
+/// Missing rows become an explicit consumer gap, not an automatic tail reset.
+pub fn prune_outbox(
+    tx: &Transaction<'_>,
+    workspace: &str,
+    through: i64,
+    now: i64,
+) -> Result<usize> {
+    let cutoff = chrono::DateTime::from_timestamp_millis(now - RETENTION_MS)
+        .ok_or_else(|| failure(PluginHostErrorCode::InvalidInput))?
+        .to_rfc3339();
+    let removed:Option<i64>=tx.query_row("SELECT MAX(sequence) FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence<=?2 AND occurred_at<?3",params![workspace,through,cutoff],|r|r.get(0))?;
+    let Some(removed) = removed else { return Ok(0) };
+    // Never create a hole past a newer retained event in the same Workspace.
+    let blocked:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence<=?2 AND occurred_at>=?3)",params![workspace,removed,cutoff],|r|r.get(0))?;
+    if blocked {
+        return Err(failure(PluginHostErrorCode::Conflict));
+    }
+    let count = tx.execute(
+        "DELETE FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND sequence<=?2",
+        params![workspace, removed],
+    )?;
+    tx.execute("INSERT INTO plugin_service_retention(workspace_id,removed_through) VALUES(?1,?2) ON CONFLICT(workspace_id) DO UPDATE SET removed_through=MAX(removed_through,excluded.removed_through)",params![workspace,removed])?;
+    Ok(count)
+}
+
+#[cfg(test)]
+#[path = "plugin_service_tests.rs"]
+mod tests;

crates/workspace-server/src/plugin_service_migration_tests.rs

@@ -0,0 +1,53 @@
+#[test]
+fn service_migration_preserves_history_has_no_backfill_and_matches_fresh() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("server.db");
+    prepare_retained_schema(&path, 111);
+    let conn = Connection::open(&path).unwrap();
+    configure_sqlite(&conn).unwrap();
+    seed_migration_workspace(&conn);
+    let before = workspace_schema_migration_history(&conn).unwrap();
+    migrate_plugin_services_v111_to_v112(&conn).unwrap();
+    assert_eq!(
+        &workspace_schema_migration_history(&conn).unwrap()[..before.len()],
+        before.as_slice()
+    );
+    assert_eq!(
+        conn.query_row(
+            "SELECT count(*) FROM typed_ticket_service_outbox",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        0
+    );
+    let fresh = SqliteWorkspaceStore::in_memory_for_test().unwrap();
+    let objects = |c: &Connection| -> Result<Vec<(String, String)>> {
+        Ok(c.prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL AND (name LIKE 'plugin_service_%' OR name LIKE 'typed_ticket_service_outbox%') ORDER BY name")?.query_map([],|r|Ok((r.get(0)?,r.get(1)?)))?.collect::<rusqlite::Result<_>>()?)
+    };
+    assert_eq!(objects(&conn).unwrap(), fresh.with_conn(objects).unwrap());
+    assert!(migrate_plugin_services_v111_to_v112(&conn).is_err());
+}
+
+#[test]
+fn service_migration_late_failure_rolls_back_domain_and_consumer_schema() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("server.db");
+    prepare_retained_schema(&path, 111);
+    let conn = Connection::open(path).unwrap();
+    configure_sqlite(&conn).unwrap();
+    conn.execute_batch("CREATE TABLE plugin_service_recovery_audit(unrelated INTEGER)")
+        .unwrap();
+    let schema = |c: &Connection| {
+        c.query_row("SELECT group_concat(sql,char(10)) FROM (SELECT sql FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY name)",[],|r|r.get::<_,String>(0)).unwrap()
+    };
+    let before = schema(&conn);
+    let history = workspace_schema_migration_history(&conn).unwrap();
+    assert!(migrate_plugin_services_v111_to_v112(&conn).is_err());
+    assert_eq!(schema(&conn), before);
+    assert_eq!(workspace_schema_migration_history(&conn).unwrap(), history);
+    conn.execute_batch("DROP TABLE plugin_service_recovery_audit")
+        .unwrap();
+    migrate_plugin_services_v111_to_v112(&conn).unwrap();
+    assert_eq!(current_schema_version(&conn).unwrap(), 112);
+}

crates/workspace-server/src/plugin_service_schema.sql

@@ -0,0 +1,67 @@
+CREATE TABLE plugin_service_consumers (
+ workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+ consumer_id TEXT NOT NULL,
+ runtime_id TEXT NOT NULL,
+ worker_id TEXT NOT NULL,
+ installation_id TEXT NOT NULL,
+ plan_id TEXT NOT NULL,
+ service_name TEXT NOT NULL,
+ subscriptions_json TEXT NOT NULL CHECK(json_valid(subscriptions_json) AND json_type(subscriptions_json)='array' AND json_array_length(subscriptions_json) BETWEEN 1 AND 2),
+ cursor INTEGER NOT NULL CHECK(cursor>=0),
+ -- Explicit recovery may drain the already observed backlog, one event at a time.
+ -- New backlog beyond this watermark is subject to ordinary backpressure again.
+ drain_through INTEGER NOT NULL DEFAULT 0 CHECK(drain_through>=0),
+ lease_token TEXT,
+ lease_until_ms INTEGER NOT NULL DEFAULT 0,
+ pause_reason TEXT CHECK(pause_reason IN ('rejected','attempts_exhausted','queue_full','retention_gap','indeterminate','authorization')),
+ current_sequence INTEGER,
+ current_event_id TEXT,
+ invocation_id TEXT,
+ attempt INTEGER NOT NULL DEFAULT 0 CHECK(attempt BETWEEN 0 AND 5),
+ next_attempt_ms INTEGER NOT NULL DEFAULT 0,
+ last_ack_event_id TEXT,
+ PRIMARY KEY(workspace_id,consumer_id),
+ UNIQUE(workspace_id,runtime_id,worker_id,installation_id,plan_id,service_name),
+ FOREIGN KEY(workspace_id,plan_id) REFERENCES plugin_execution_plans(workspace_id,plan_id),
+ FOREIGN KEY(workspace_id,installation_id) REFERENCES plugin_installations(workspace_id,installation_id),
+ CHECK((current_sequence IS NULL AND current_event_id IS NULL AND invocation_id IS NULL) OR (current_sequence IS NOT NULL AND current_event_id IS NOT NULL))
+);
+-- Release proofs outlive lease expiry/rotation: a lost response can be retried
+-- for the lifetime of the consumer. Only deleting that owner may discard proofs.
+CREATE TABLE plugin_service_released_proofs (
+ workspace_id TEXT NOT NULL,
+ consumer_id TEXT NOT NULL,
+ lease_token TEXT NOT NULL CHECK(length(CAST(lease_token AS BLOB)) BETWEEN 1 AND 512),
+ released_at_ms INTEGER NOT NULL,
+ PRIMARY KEY(workspace_id,consumer_id,lease_token),
+ FOREIGN KEY(workspace_id,consumer_id) REFERENCES plugin_service_consumers(workspace_id,consumer_id) ON DELETE CASCADE
+);
+CREATE TRIGGER plugin_service_released_proof_immutable BEFORE UPDATE ON plugin_service_released_proofs
+BEGIN SELECT RAISE(ABORT,'Service released proof is immutable'); END;
+CREATE TRIGGER plugin_service_released_proof_retained BEFORE DELETE ON plugin_service_released_proofs
+WHEN EXISTS(SELECT 1 FROM plugin_service_consumers WHERE workspace_id=OLD.workspace_id AND consumer_id=OLD.consumer_id)
+BEGIN SELECT RAISE(ABORT,'Service released proof must be retained'); END;
+CREATE TABLE plugin_service_retention (
+ workspace_id TEXT PRIMARY KEY REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+ removed_through INTEGER NOT NULL CHECK(removed_through>=0)
+);
+CREATE TABLE plugin_service_recovery_audit (
+ workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+ operation_key TEXT NOT NULL CHECK(length(CAST(operation_key AS BLOB)) BETWEEN 1 AND 128),
+ consumer_id TEXT NOT NULL,
+ fingerprint TEXT NOT NULL,
+ actor TEXT NOT NULL,
+ reason TEXT NOT NULL CHECK(length(CAST(reason AS BLOB)) BETWEEN 1 AND 1024),
+ action_json TEXT NOT NULL CHECK(json_valid(action_json)),
+ at_ms INTEGER NOT NULL,
+ PRIMARY KEY(workspace_id,operation_key),
+ FOREIGN KEY(workspace_id,consumer_id) REFERENCES plugin_service_consumers(workspace_id,consumer_id)
+);
+CREATE TRIGGER plugin_service_identity_immutable BEFORE UPDATE OF workspace_id,consumer_id,runtime_id,worker_id,installation_id,plan_id,service_name,subscriptions_json ON plugin_service_consumers
+BEGIN SELECT RAISE(ABORT,'Service consumer identity is immutable'); END;
+CREATE TRIGGER plugin_service_cursor_forward BEFORE UPDATE OF cursor ON plugin_service_consumers WHEN NEW.cursor<OLD.cursor
+BEGIN SELECT RAISE(ABORT,'Service cursor cannot rewind'); END;
+CREATE TRIGGER plugin_service_audit_immutable BEFORE UPDATE ON plugin_service_recovery_audit
+BEGIN SELECT RAISE(ABORT,'Service recovery audit is immutable'); END;
+CREATE TRIGGER plugin_service_audit_retained BEFORE DELETE ON plugin_service_recovery_audit WHEN EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=OLD.workspace_id)
+BEGIN SELECT RAISE(ABORT,'Service recovery audit must be retained'); END;

crates/workspace-server/src/plugin_settings.rs

@@ -45,6 +45,10 @@ pub struct PluginEnablement {
     pub package_digest: String,
     pub artifact_digest: String,
     pub tools: Vec<String>,
+    #[serde(default)]
+    pub services: Vec<String>,
+    #[serde(default)]
+    pub ticket_events: bool,
     pub logging: bool,
     #[serde(default)]
     pub ticket_read: bool,
@@ -170,7 +174,7 @@ pub(crate) fn evaluate_catalog_config(
     // Each alias has its own closed schema. Reusing an artifact with distinct
     // bindings is intentional and does not manufacture another schema authority.
     let enabled_fields = artifacts.iter().map(|(alias, artifact)| {
-        format!("{} = {{ installation_id = String; expected_last_operation = String; package_digest = String; artifact_digest = String; tools = [...String]; logging = Bool default false; ticket_read = Bool default false; ticket_data_read = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; ticket_data_write = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; connections = {{ ...String }} default {{}}; config = ({}); }};", alias, artifact.config_schema.dcdl)
+        format!("{} = {{ installation_id = String; expected_last_operation = String; package_digest = String; artifact_digest = String; tools = [...String]; services = [...String] default []; ticket_events = Bool default false; logging = Bool default false; ticket_read = Bool default false; ticket_data_read = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; ticket_data_write = [...{{ collection = String; schema_id = String; schema_version = String; }}] default []; connections = {{ ...String }} default {{}}; config = ({}); }};", alias, artifact.config_schema.dcdl)
     }).collect::<Vec<_>>().join("\n");
     let enabled_schema = if enabled_fields.is_empty() {
         "{} default {}".to_string()
@@ -245,6 +249,7 @@ fn validate_enablement(
                 | manifest::plugin::PLUGIN_COMPONENT_LOGGED_TOOL_WORLD
                 | manifest::plugin::PLUGIN_COMPONENT_HTTPS_LOGGED_TOOL_WORLD
                 | "yoi:plugin/ticket-tool-plugin@2.1.0"
+                | "yoi:plugin/ticket-service-plugin@2.2.0"
         )
         || artifact.required_imports.iter().any(|i| {
             !matches!(
@@ -269,6 +274,21 @@ fn validate_enablement(
     {
         return Err(config_error("unknown or duplicate Tool selection"));
     }
+    if enabled.services.iter().collect::<BTreeSet<_>>().len() != enabled.services.len()
+        || enabled
+            .services
+            .iter()
+            .any(|name| !artifact.services.iter().any(|s| &s.name == name))
+        || (enabled.ticket_events
+            && (!artifact
+                .requested_grants
+                .contains(&PluginGrantMetadata::TicketEvents)
+                || artifact.world != "yoi:plugin/ticket-service-plugin@2.2.0"))
+    {
+        return Err(config_error(
+            "Service selection/grant exceeds package request",
+        ));
+    }
     if enabled.logging
         && !artifact
             .requested_grants
@@ -432,6 +452,11 @@ fn project_profile_plugins(
             .tools
             .iter()
             .any(|tool| !enabled.tools.contains(tool))
+            || selection
+                .services
+                .iter()
+                .any(|s| !enabled.services.contains(s))
+            || (selection.ticket_events && !enabled.ticket_events)
             || (selection.logging && !enabled.logging)
             || (selection.ticket_read && !enabled.ticket_read)
             || selection
@@ -461,6 +486,9 @@ fn project_profile_plugins(
         let config =
             evaluate_effective_config(&config, &resolved.artifacts[alias].config_schema.dcdl)?;
         let mut grants = Vec::new();
+        if selection.ticket_events {
+            grants.push(PluginGrant::TicketEvents);
+        }
         if selection.logging {
             grants.push(PluginGrant::Logging);
         }
@@ -498,12 +526,15 @@ fn project_profile_plugins(
             &resolved.artifacts[alias],
             PluginPinProjection {
                 config,
-                surfaces: if selection.tools.is_empty() {
-                    vec![]
-                } else {
-                    vec![PluginSurface::Tool]
-                },
+                surfaces: [
+                    (!selection.tools.is_empty()).then_some(PluginSurface::Tool),
+                    (!selection.services.is_empty()).then_some(PluginSurface::Service),
+                ]
+                .into_iter()
+                .flatten()
+                .collect(),
                 tools: selection.tools.clone(),
+                services: selection.services.clone(),
                 grants,
                 bindings,
             },

crates/workspace-server/src/schema_convergence_tests.rs

@@ -124,6 +124,10 @@ mod schema_convergence {
                             version: 111,
                             name: "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
                         },
+                        WorkspaceSchemaMigrationStep {
+                            version: 112,
+                            name: "Ticket Service outbox and fenced consumers".into(),
+                        },
                     ]
                 );
                 assert_eq!(
@@ -146,6 +150,7 @@ mod schema_convergence {
                     expected.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
                     expected.push((110, "Worker Plugin launch selection associations".into()));
                     expected.push((111, "Typed Plugin Ticket extension data and atomic mutation receipts".into()));
+                    expected.push((112, "Ticket Service outbox and fenced consumers".into()));
                     assert_eq!(workspace_schema_migration_history(conn)?, expected);
                     for table in ["passkey_credentials", "passkey_metadata", "passkey_addition_receipts", "backend_jobs", "backend_job_attempts"] {
                         assert_eq!(state_contract_schema_signature(conn, table), fresh.with_conn(|c| Ok(state_contract_schema_signature(c, table)))?, "{table}");

crates/workspace-server/src/server.rs

@@ -13766,6 +13766,30 @@ impl server_api::ServerApi for ServerApiContractService {
         plugin::list(api, actor, workspace_id, query).await
     }
 
+    async fn plugin_service_diagnostics(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        consumer_id: String,
+    ) -> std::result::Result<server_api::PluginServiceDiagnostics, server_api::RepositoryApiError>
+    {
+        self.workspace_api()?
+            .plugin_service_diagnostics(workspace_id, actor, consumer_id)
+            .await
+    }
+
+    async fn plugin_service_recover(
+        &self,
+        actor: RequestActor,
+        workspace_id: String,
+        request: server_api::PluginServiceRecoveryRequest,
+    ) -> std::result::Result<server_api::PluginServiceDiagnostics, server_api::RepositoryApiError>
+    {
+        self.workspace_api()?
+            .recover_plugin_service(workspace_id, actor, request)
+            .await
+    }
+
     async fn plugin_worker_diagnostics(
         &self,
         actor: RequestActor,

crates/workspace-server/src/server/subject_authorization.rs

@@ -955,7 +955,9 @@ impl WorkspaceApi {
                         "foreign Plugin Worker".into(),
                     ));
                 }
-                plugin_execution_identity(&api, &worker).map_err(|error| error.error)
+                super::plugin_host::with_authority_database(&api, || {
+                    plugin_execution_identity(&api, &worker).map_err(|error| error.error)
+                })
             })
             .await
     }

crates/workspace-server/src/server_plugin.rs

@@ -17,6 +17,8 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
         .merge(server_api_axum::plugin_install(service.clone()))
         .merge(server_api_axum::plugin_list(service.clone()))
         .merge(server_api_axum::plugin_worker_diagnostics(service.clone()))
+        .merge(server_api_axum::plugin_service_diagnostics(service.clone()))
+        .merge(server_api_axum::plugin_service_recover(service.clone()))
         .merge(server_api_axum::plugin_show(service.clone()))
         .merge(server_api_axum::plugin_artifact_add(service.clone()))
         .merge(server_api_axum::plugin_revoke(service.clone()))

crates/workspace-server/src/server_plugin_authority_storage_tests.rs

@@ -0,0 +1,198 @@
+//! Focused physical-lifetime/live-authority regressions through existing owners.
+//! Storage counters observe real directory sync/open; they never replace that I/O.
+use super::*;
+
+pub(super) fn independent_store(
+    api: &WorkspaceApi,
+) -> (crate::FeatureStorage, crate::subjektiv::SubjektivStore) {
+    let manager = crate::FeatureStorage::for_server_database(&api.config.database_path).unwrap();
+    let scope = manager.workspace(TEST_WORKSPACE_ID).unwrap();
+    let registration = crate::subjektiv::SubjektivStore::register(&scope).unwrap();
+    let store = crate::subjektiv::SubjektivStore::open(&scope, &registration).unwrap();
+    // Handles do not own manager lifetime: dropping it intentionally fences them.
+    (manager, store)
+}
+
+fn permission(f: &Fixture) -> server_api::authorization::OperationAuthorizationRequest {
+    server_api::authorization::OperationAuthorizationRequest {
+        operation: "plugin.admit".into(),
+        resource: PermissionResource::PluginInstallation {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+            installation_id: f.installed.installation.installation_id.clone(),
+        },
+    }
+}
+
+fn revoke(store: &crate::subjektiv::SubjektivStore, subject: &str) {
+    let previous = store.subject_permissions(subject).unwrap().unwrap();
+    store
+        .set_subject_permissions(
+            subject,
+            &SubjectPermissionsUpdate {
+                operation_id: "external-revoke".into(),
+                expected_change_id: Some(previous.change_id),
+                execution: vec![],
+                delegation: vec![],
+            },
+        )
+        .unwrap();
+}
+
+#[tokio::test]
+async fn sync_authority_checks_reduce_physical_initializations_without_skipping_live_reads() {
+    let f = Fixture::new(None).await;
+    // Call the unchanged synchronous owners both without and with physical-handle
+    // retention. These are the identity and dual-permission portions of Host checks.
+    for identity in [true, false] {
+        let check = || {
+            if identity {
+                subject_authorization::plugin_execution_identity(&f.api, &f.source)
+                    .map(|_| ())
+                    .map_err(|e| e.error)
+            } else {
+                for operation in ["plugin.service.consume", "plugin.admit"] {
+                    let mut request = permission(&f);
+                    request.operation = operation.into();
+                    subject_authorization::authorize_scope(&f.api, &f.source, &request)
+                        .map_err(|e| e.error)?;
+                }
+                Ok(())
+            }
+        };
+        let before = f.api.subjektiv_registration.open_observation();
+        check().unwrap();
+        let unretained = f
+            .api
+            .subjektiv_registration
+            .open_observation()
+            .since(before);
+        assert!(unretained.physical_opens > 1, "{unretained:?}");
+        assert_check_physical_initializations(&f, before, unretained.physical_opens);
+        let before = f.api.subjektiv_registration.open_observation();
+        plugin_host::with_authority_database(&f.api, check).unwrap();
+        assert_check_physical_initializations(&f, before, 1);
+        let retained = f
+            .api
+            .subjektiv_registration
+            .open_observation()
+            .since(before);
+        println!(
+            "authority_physical_io identity={identity} before={unretained:?} after={retained:?}"
+        );
+    }
+    f.shutdown().await;
+}
+
+#[tokio::test]
+async fn scoped_subject_database_rereads_policy_changed_on_independent_connection() {
+    let f = Fixture::new(None).await;
+    let (_manager, other) = independent_store(&f.api);
+    let request = permission(&f);
+    let before = f.api.subjektiv_registration.open_observation();
+    plugin_host::with_authority_database(&f.api, || {
+        subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap();
+        revoke(&other, &f.subject);
+        let error =
+            subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap_err();
+        assert!(matches!(error.error, Error::WorkspacePermissionDenied(_)));
+        Ok(())
+    })
+    .unwrap();
+    assert_check_physical_initializations(&f, before, 1);
+    drop(other);
+    f.shutdown().await;
+}
+
+#[tokio::test]
+async fn scoped_subject_database_does_not_survive_deletion_or_refresh_itself_after_restore() {
+    let f = Fixture::new(None).await;
+    let snapshot_dir = tempfile::tempdir().unwrap();
+    let snapshot = snapshot_dir.path().join("snapshot");
+    f.api.feature_storage.backup(&snapshot).unwrap();
+    let request = permission(&f);
+    plugin_host::with_authority_database(&f.api, || {
+        subject_authorization::authorize_scope(&f.api, &f.source, &request).unwrap();
+        // The guard retains no connection lock: deletion completes synchronously.
+        f.api.feature_storage.delete().unwrap();
+        assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
+        assert!(f.api.feature_storage.restore(&snapshot).is_err());
+        let fresh =
+            crate::FeatureStorage::for_server_database(&f.api.config.database_path).unwrap();
+        let scope = fresh.workspace(TEST_WORKSPACE_ID).unwrap();
+        scope.restore(&snapshot).unwrap();
+        let registration = crate::subjektiv::SubjektivStore::register(&scope).unwrap();
+        let mut restored = f.api.clone();
+        restored.feature_storage = scope;
+        restored.subjektiv_registration = registration;
+        // A fresh Host binding can read the restored DB; the old binding stays fenced.
+        subject_authorization::authorize_scope(&restored, &f.source, &request).unwrap();
+        assert!(subject_authorization::authorize_scope(&f.api, &f.source, &request).is_err());
+        restored.feature_storage.shutdown().unwrap();
+        Ok(())
+    })
+    .unwrap();
+    f.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn cancelled_authority_waiter_keeps_physical_database_until_accepted_check_finishes() {
+    let f = Fixture::new(None).await;
+    let (_manager, other) = independent_store(&f.api);
+    let request = permission(&f);
+    let executor = f.api.config_store.db_executor();
+    let api = f.api.clone();
+    let worker = f.source.clone();
+    let (entered, started) = tokio::sync::oneshot::channel();
+    let (release, released) = std::sync::mpsc::sync_channel(0);
+    let denied = Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let checked = denied.clone();
+    let lane = executor.clone();
+    let before = f.api.subjektiv_registration.open_observation();
+    let waiter = tokio::spawn(async move {
+        lane.run(move || {
+            plugin_host::with_authority_database(&api, || {
+                subject_authorization::authorize_scope(&api, &worker, &request).unwrap();
+                entered.send(()).unwrap();
+                released.recv().unwrap();
+                checked.store(
+                    subject_authorization::authorize_scope(&api, &worker, &request).is_err(),
+                    std::sync::atomic::Ordering::SeqCst,
+                );
+                Ok(())
+            })
+        })
+        .await
+    });
+    tokio::time::timeout(std::time::Duration::from_secs(10), started)
+        .await
+        .unwrap()
+        .unwrap();
+    waiter.abort();
+    assert!(waiter.await.unwrap_err().is_cancelled());
+    // This open reuses the accepted job's physical handle even after its waiter dies.
+    drop(
+        f.api
+            .feature_storage
+            .open(&f.api.subjektiv_registration)
+            .unwrap(),
+    );
+    assert_check_physical_initializations(&f, before, 1);
+    revoke(&other, &f.subject);
+    release.send(()).unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(10), executor.drain())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(denied.load(std::sync::atomic::Ordering::SeqCst));
+    assert_check_physical_initializations(&f, before, 1);
+    // Draining the accepted work released the guard, rather than creating a cache.
+    drop(
+        f.api
+            .feature_storage
+            .open(&f.api.subjektiv_registration)
+            .unwrap(),
+    );
+    assert_check_physical_initializations(&f, before, 2);
+    drop(other);
+    f.shutdown().await;
+}

crates/workspace-server/src/server_plugin_controller_tests.rs

@@ -1,6 +1,7 @@
 // Include directly in server::tests (this file owns its module).
 // Production Embedded Server assembly -> ProfileRuntimeWorkerFactory -> actual
-// Controller/Engine -> pinned Wasmtime Tool -> HTTPS -> SQLite and Session store.
+// Controller/Engine -> pinned Wasmtime Tool/Service -> HTTPS, Ticket outbox,
+// SQLite receipts/cursors and Session store.
 // Only the external model and the TLS peer's DNS/socket/trust are substituted.
 // No FixtureRuntime, manually prepared Plugin Feature, or fabricated Host owner.
 //
@@ -31,6 +32,12 @@ mod plugin_controller_tests {
     const TOOL: &str = "example_status";
     const WATCHDOG: Duration = Duration::from_secs(60);
 
+    enum FixtureSurface {
+        Tool,
+        HttpsTool,
+        Service,
+    }
+
     struct Fixture {
         api: WorkspaceApi,
         owner: String,
@@ -113,23 +120,42 @@ mod plugin_controller_tests {
             &self,
             peer: &Peer,
             https: bool,
+        ) -> (server_api::PluginOperationResponse, Vec<u8>) {
+            self.install_surface(
+                peer,
+                if https {
+                    FixtureSurface::HttpsTool
+                } else {
+                    FixtureSurface::Tool
+                },
+            )
+            .await
+        }
+
+        async fn install_surface(
+            &self,
+            peer: &Peer,
+            surface: FixtureSurface,
         ) -> (server_api::PluginOperationResponse, Vec<u8>) {
             use manifest::plugin::*;
             let directory = tempfile::tempdir().unwrap();
-            let (manifest, component): (&str, &[u8]) = if https {
-                (
+            let (manifest, component): (&str, &[u8]) = match surface {
+                FixtureSurface::Service => (
+                    include_str!("../../plugin-runtime/tests/service-guest/plugin.toml"),
+                    include_bytes!("../tests/fixtures/ticket-service-v2.component.wasm"),
+                ),
+                FixtureSurface::HttpsTool => (
                     include_str!(
                         "../../../resources/plugin/templates/rust-component-https/plugin.toml"
                     ),
                     include_bytes!("../../plugin-pdk/tests/fixtures/https-tool-v2.component.wasm"),
-                )
-            } else {
-                (
+                ),
+                FixtureSurface::Tool => (
                     include_str!(
                         "../../../resources/plugin/templates/rust-component-tool/plugin.toml"
                     ),
                     include_bytes!("../../plugin-pdk/tests/fixtures/tool-v2.component.wasm"),
-                )
+                ),
             };
             fs::write(directory.path().join("plugin.toml"), manifest).unwrap();
             fs::write(directory.path().join("plugin.component.wasm"), component).unwrap();
@@ -168,11 +194,10 @@ mod plugin_controller_tests {
                 })).unwrap(), false, StatusCode::OK).await;
             let installed: server_api::PluginOperationResponse =
                 serde_json::from_value(installed).unwrap();
-            *self.model_tool.lock().unwrap() = format!(
-                "{}__{}",
-                installed.installation.installation_id,
-                installed.installation.artifacts[0].tools[0].name
-            );
+            if let Some(tool) = installed.installation.artifacts[0].tools.first() {
+                *self.model_tool.lock().unwrap() =
+                    format!("{}__{}", installed.installation.installation_id, tool.name,);
+            }
             self.configure(&installed, peer, true).await;
             (installed, package.bytes)
         }
@@ -187,18 +212,26 @@ mod plugin_controller_tests {
             let mut binding = peer.binding();
             binding.methods = vec![server_api::PluginHttpMethod::Get];
             binding.path_prefixes = vec!["/status".into()];
-            let tool = &artifact.tools[0].name;
-            let https = tool == TOOL;
+            let service = artifact.services.first().map(|service| &service.name);
+            let tool = artifact.tools.first().map(|tool| &tool.name);
+            let https = tool.is_some_and(|tool| tool == TOOL);
+            let tools = tool.into_iter().collect::<Vec<_>>();
+            let services = service.into_iter().collect::<Vec<_>>();
+            let config = if service.is_some() {
+                json!({"mode":"normal"})
+            } else {
+                json!({})
+            };
             let plugins = json!({"enabled":{"controller":{
                 "installation_id":installed.installation.installation_id,
                 "expected_last_operation":installed.operation_id,
                 "package_digest":artifact.package_digest,"artifact_digest":artifact.artifact_digest,
-                "tools":[tool],"logging":false,"config":{},
+                "tools":tools,"services":services,"ticket_events":service.is_some(),"logging":service.is_some(),"config":config,
                 "connections":if https { json!({"api":"remote"}) } else { json!({}) }
             }},"bindings":if https { json!({"remote":binding}) } else { json!({}) }});
             let selection = if selected {
                 json!({"selections":[{"installation_id":installed.installation.installation_id,
-                    "tools":[tool],"logging":false,"https_slots":if https { vec!["api"] } else { vec![] }}]})
+                    "tools":tools,"services":services,"ticket_events":service.is_some(),"logging":service.is_some(),"https_slots":if https { vec!["api"] } else { vec![] }}]})
             } else {
                 json!({"selections":[]})
             };
@@ -552,7 +585,12 @@ mod plugin_controller_tests {
             peer.connection_count(),
             2,
             "a new logical call must not replay the first receipt; model requests: {:?}",
-            f.requests.lock().unwrap().iter().map(|r| &r.items).collect::<Vec<_>>()
+            f.requests
+                .lock()
+                .unwrap()
+                .iter()
+                .map(|r| &r.items)
+                .collect::<Vec<_>>()
         );
         assert_eq!(f.count("plugin_invocations"), 2);
         assert_eq!(f.count("plugin_effect_receipts"), 2);
@@ -742,6 +780,310 @@ mod plugin_controller_tests {
         f.close().await;
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn embedded_service_audited_recovery_wakes_same_controller_and_restore_keeps_cursor() {
+        use crate::plugin_service::Consumer;
+        use rusqlite::OptionalExtension;
+        use server_api::{PluginServiceDiagnostics, PluginServicePauseReason};
+
+        async fn until(
+            phase: &str,
+            observe: impl Fn() -> Option<Consumer>,
+            ready: impl Fn(&Consumer) -> bool,
+        ) -> Consumer {
+            tokio::time::timeout(WATCHDOG, async {
+                loop {
+                    if let Some(consumer) = observe().filter(&ready) {
+                        return consumer;
+                    }
+                    tokio::time::sleep(Duration::from_millis(20)).await;
+                }
+            })
+            .await
+            .unwrap_or_else(|_| {
+                let state = observe().map(|c| {
+                    (
+                        c.consumer_id,
+                        c.cursor,
+                        c.pause,
+                        c.current_event_id,
+                        c.attempt,
+                        c.next_attempt_ms,
+                        c.lease_token.is_some(),
+                    )
+                });
+                panic!("Service {phase} did not converge: {state:?}");
+            })
+        }
+
+        let peer = Peer::new(HOST, vec![]).await;
+        let f = Fixture::new(&peer).await;
+        let (installed, _) = f.install_surface(&peer, FixtureSurface::Service).await;
+        let worker = f.create(PROFILE).await;
+        let state = || {
+            f.api.config_store.with_conn(|conn| {
+                let id: Option<String> = conn.query_row(
+                    "SELECT consumer_id FROM plugin_service_consumers WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",
+                    rusqlite::params![TEST_WORKSPACE_ID, worker.runtime_id, worker.worker_id],
+                    |row| row.get(0),
+                ).optional()?;
+                id.map(|id| crate::plugin_service::load(conn, TEST_WORKSPACE_ID, &id)).transpose()
+            }).unwrap()
+        };
+        let before = until("initial factory lease", &state, |c| c.lease_token.is_some()).await;
+        let pin = f.execution_request(&worker)["plugin_execution"].clone();
+        assert_eq!(pin["plan_id"], before.plan_id);
+        assert_eq!(f.model_clients.load(Ordering::SeqCst), 1);
+        let router = build_router(f.api.clone());
+        let prefix = format!("/api/w/{TEST_WORKSPACE_ID}");
+        let path = format!(
+            "{prefix}/plugins/worker-diagnostics/{}/{}",
+            worker.runtime_id, worker.worker_id
+        );
+        let observed = request(
+            router.clone(),
+            Some(&f.owner),
+            "GET",
+            &path,
+            vec![],
+            false,
+            StatusCode::OK,
+        )
+        .await;
+        let observed: server_api::PluginWorkerDiagnostics =
+            serde_json::from_value(observed).unwrap();
+        let server_api::PluginWorkerDiagnosticSelection::Plan {
+            plan_id, plugins, ..
+        } = observed.selection
+        else {
+            panic!("sealed Service launch selection missing");
+        };
+        assert_eq!(plan_id, before.plan_id);
+        assert_eq!(plugins[0].selected_services, ["ticket_events"]);
+        assert_eq!(
+            plugins[0].service_consumers[0].diagnostics.consumer_id,
+            before.consumer_id
+        );
+        assert_eq!(
+            plugins[0].admission,
+            server_api::PluginDiagnosticAdmission::Admissible
+        );
+        let consumer_path = format!("{prefix}/plugins/service-consumers/{}", before.consumer_id);
+        let recover_path = format!("{prefix}/plugins/service-consumers/recover");
+        request(
+            router.clone(),
+            None,
+            "GET",
+            &consumer_path,
+            vec![],
+            false,
+            StatusCode::UNAUTHORIZED,
+        )
+        .await;
+        let observed = request(
+            router.clone(),
+            Some(&f.owner),
+            "GET",
+            &consumer_path,
+            vec![],
+            false,
+            StatusCode::OK,
+        )
+        .await;
+        assert_eq!(observed["consumer_id"], before.consumer_id);
+        assert_eq!(observed["lease_active"], true);
+        assert!(!observed.to_string().contains("lease_token"));
+        request(router.clone(), Some(&f.owner), "POST", &recover_path, serde_json::to_vec(&json!({
+            "consumer_id":before.consumer_id,"expected_pause":"rejected","action":{"kind":"resume"},"operation_key":"not-paused","reason":"must not override an active consumer"
+        })).unwrap(), false, StatusCode::CONFLICT).await;
+        let unchanged = state().unwrap();
+        assert_eq!(unchanged.cursor, before.cursor);
+        assert_eq!(unchanged.lease_token, before.lease_token);
+        assert!(unchanged.pause.is_none());
+        assert_eq!(f.count("plugin_invocations"), 0);
+        assert_eq!(peer.connection_count(), 0);
+        assert!(f.requests.lock().unwrap().is_empty());
+        assert_eq!(
+            installed.installation.artifacts[0].services[0].name,
+            "ticket_events"
+        );
+
+        // The real guest requests extension_write, which this events/logging-only
+        // selection denies. Its RetryAfter exhausts the durable delivery budget.
+        let ticket = browser_ticket_backend(&f.api)
+            .unwrap()
+            .create(ticket::NewTicket::new("Same Controller Service recovery"))
+            .unwrap();
+        let exhausted = until("initial attempts exhausted and released", &state, |c| {
+            c.pause == Some(PluginServicePauseReason::AttemptsExhausted) && c.lease_token.is_none()
+        })
+        .await;
+        let observed = request(
+            router.clone(),
+            Some(&f.owner),
+            "GET",
+            &consumer_path,
+            vec![],
+            false,
+            StatusCode::OK,
+        )
+        .await;
+        let paused: PluginServiceDiagnostics = serde_json::from_value(observed).unwrap();
+        assert_eq!(paused.consumer_id, before.consumer_id);
+        assert_eq!(
+            paused.pause_reason,
+            Some(PluginServicePauseReason::AttemptsExhausted)
+        );
+        assert!(!paused.lease_active);
+        assert_eq!(paused.attempt, crate::plugin_service::MAX_ATTEMPTS);
+        assert_eq!(paused.current_event_id, exhausted.current_event_id);
+        let event_id = paused.current_event_id.as_ref().unwrap();
+        let sequence: i64 = f.api.config_store.with_conn(|conn| {
+            Ok(conn.query_row(
+                "SELECT sequence FROM typed_ticket_service_outbox WHERE workspace_id=?1 AND event_id=?2 AND ticket_id=?3",
+                rusqlite::params![TEST_WORKSPACE_ID, event_id, ticket.id.as_str()],
+                |row| row.get(0),
+            )?)
+        }).unwrap();
+        assert!(sequence > before.cursor);
+        assert_eq!(exhausted.cursor, before.cursor);
+        assert!(exhausted.next_attempt_ms > 0);
+        assert!(exhausted.invocation_id.is_some());
+        assert_eq!(f.count("plugin_invocations"), 1);
+
+        let resume_reason = "Retry the observed denied event on its existing Controller";
+        let resumed = request(router.clone(), Some(&f.owner), "POST", &recover_path, serde_json::to_vec(&json!({
+            "consumer_id":paused.consumer_id,"expected_pause":paused.pause_reason.unwrap(),
+            "action":{"kind":"resume"},"operation_key":"controller-resume","reason":resume_reason
+        })).unwrap(), false, StatusCode::OK).await;
+        assert_eq!(resumed["consumer_id"], before.consumer_id);
+        assert!(resumed["pause_reason"].is_null());
+        assert_eq!(resumed["current_event_id"], event_id.as_str());
+
+        // No stop/restore, Submit, manual Acquire, or replacement owner occurs
+        // here. A later durable retry deadline proves new guest delivery rather
+        // than accidentally accepting the already exhausted pre-recovery state.
+        let retried = until(
+            "same Controller recovery and renewed exhaustion",
+            &state,
+            |c| {
+                c.consumer_id == before.consumer_id
+                    && c.current_event_id.as_ref() == Some(event_id)
+                    && c.pause == paused.pause_reason
+                    && c.lease_token.is_none()
+                    && c.next_attempt_ms > exhausted.next_attempt_ms
+            },
+        )
+        .await;
+        assert_eq!(retried.plan_id, before.plan_id);
+        assert_eq!(retried.cursor, before.cursor);
+        assert_eq!(retried.attempt, crate::plugin_service::MAX_ATTEMPTS);
+        assert_eq!(retried.invocation_id, exhausted.invocation_id);
+        assert_eq!(f.count("plugin_invocations"), 1);
+        assert_eq!(f.model_clients.load(Ordering::SeqCst), 1);
+        let observed = request(
+            router.clone(),
+            Some(&f.owner),
+            "GET",
+            &consumer_path,
+            vec![],
+            false,
+            StatusCode::OK,
+        )
+        .await;
+        let paused_again: PluginServiceDiagnostics = serde_json::from_value(observed).unwrap();
+        assert_eq!(paused_again, paused);
+
+        let skip_reason = "Explicitly abandon the observed denied queue head";
+        let skipped = request(router.clone(), Some(&f.owner), "POST", &recover_path, serde_json::to_vec(&json!({
+            "consumer_id":paused_again.consumer_id,"expected_pause":paused_again.pause_reason.unwrap(),
+            "action":{"kind":"skip","event_id":event_id},"operation_key":"controller-skip","reason":skip_reason
+        })).unwrap(), false, StatusCode::OK).await;
+        assert_eq!(skipped["consumer_id"], before.consumer_id);
+        assert!(skipped["pause_reason"].is_null());
+        assert!(skipped["current_event_id"].is_null());
+        let checkpoint = until(
+            "same Controller reacquisition after audited Skip",
+            &state,
+            |c| {
+                c.consumer_id == before.consumer_id
+                    && c.cursor == sequence
+                    && c.pause.is_none()
+                    && c.current_event_id.is_none()
+                    && c.lease_token.is_some()
+            },
+        )
+        .await;
+        assert!(checkpoint.cursor > 0);
+        assert_eq!(checkpoint.plan_id, before.plan_id);
+        assert_ne!(checkpoint.lease_token, before.lease_token);
+        assert_eq!(f.model_clients.load(Ordering::SeqCst), 1);
+
+        let owner = f
+            .api
+            .store
+            .get_workspace(TEST_WORKSPACE_ID)
+            .await
+            .unwrap()
+            .unwrap()
+            .owner_account_id;
+        let audit: Vec<(String, String, String, String, String)> = f.api.config_store.with_conn(|conn| {
+            Ok(conn.prepare(
+                "SELECT operation_key,consumer_id,actor,reason,action_json FROM plugin_service_recovery_audit WHERE workspace_id=?1 ORDER BY operation_key",
+            )?.query_map([TEST_WORKSPACE_ID], |row| Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?)))?
+                .collect::<rusqlite::Result<_>>()?)
+        }).unwrap();
+        assert_eq!(
+            audit.len(),
+            2,
+            "the rejected no-op must not create an audit"
+        );
+        for (row, (key, reason, action)) in audit.iter().zip([
+            ("controller-resume", resume_reason, json!({"kind":"resume"})),
+            (
+                "controller-skip",
+                skip_reason,
+                json!({"kind":"skip","event_id":event_id}),
+            ),
+        ]) {
+            assert_eq!(row.0, key);
+            assert_eq!(row.1, before.consumer_id);
+            assert_eq!(
+                serde_json::from_str::<Value>(&row.2).unwrap(),
+                json!([owner, "user-plugin-controller-owner"])
+            );
+            assert_eq!(row.3, reason);
+            assert_eq!(serde_json::from_str::<Value>(&row.4).unwrap(), action);
+        }
+
+        f.stop(&worker).await;
+        assert!(state().unwrap().lease_token.is_none());
+        f.restore(&worker, "restore-service").await;
+        let restored = until("restored factory lease", &state, |c| {
+            c.lease_token.is_some()
+        })
+        .await;
+        assert_eq!(restored.consumer_id, checkpoint.consumer_id);
+        assert_eq!(restored.plan_id, checkpoint.plan_id);
+        assert_eq!(restored.cursor, checkpoint.cursor);
+        assert!(restored.pause.is_none());
+        assert!(restored.current_event_id.is_none());
+        assert_ne!(restored.lease_token, checkpoint.lease_token);
+        assert_eq!(f.execution_request(&worker)["plugin_execution"], pin);
+        let retained = f
+            .api
+            .config_store
+            .plugin_plan_load(TEST_WORKSPACE_ID, &restored.plan_id)
+            .unwrap();
+        assert_eq!(retained.pin_state, PlanPinState::Active);
+        assert_eq!(f.count("plugin_invocations"), 1);
+        assert_eq!(f.count("plugin_effect_receipts"), 0);
+        assert_eq!(peer.connection_count(), 0);
+        assert!(f.requests.lock().unwrap().is_empty());
+        f.stop(&worker).await;
+        f.close().await;
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn embedded_builtin_create_uses_profile_factory_and_seals_empty_plugin_selection() {
         let peer = Peer::new(HOST, vec![]).await;

crates/workspace-server/src/server_plugin_data.rs

@@ -66,7 +66,7 @@ fn access_granted(admitted: &policy::AdmittedPluginOperation, access: PluginTick
         })
 }
 
-fn authorize_ticket(
+pub(super) fn authorize_ticket(
     api: &WorkspaceApi,
     context: &ServerRequestContext,
     workspace: &str,
@@ -268,11 +268,12 @@ pub(super) async fn execute(
     let config = api.config_store.clone();
     let w = workspace.clone();
     let id = invocation_id.clone();
-    let record = config
+    let mut record = config
         .db_executor()
         .run(move || config.with_conn(|conn| receipt::lookup(conn, &w, &id)))
         .await
         .map_err(policy::map_error)?;
+    record.service_lease = service_lease();
     let admitted = check(
         api.clone(),
         context.clone(),
@@ -298,8 +299,10 @@ pub(super) async fn execute(
         let ticket_id = authorize_ticket(&api, &context, &workspace, &selector, action.access())?;
         config.with_immediate_transaction(|tx| {
             let record = receipt::verify_record(tx, &record)?;
-            let tools = match &record.consumer { PluginConsumer::ToolCall { tool_name, .. } => vec![tool_name.clone()], _ => return Err(policy::failure(PluginHostErrorCode::Unsupported)) };
-            let admitted = policy::authorize(tx, &workspace, &record.execution, &record.plan_id, &record.installation_id, &tools, &PluginHostOperation::Admit)?;
+            let admitted = match &record.consumer {
+                PluginConsumer::ToolCall { tool_name, .. } => policy::authorize(tx, &workspace, &record.execution, &record.plan_id, &record.installation_id, &[tool_name.clone()], &PluginHostOperation::Admit)?,
+                PluginConsumer::Consumer {export_name, ..} => policy::authorize_service(tx, &workspace, &record.execution, &record.plan_id, &record.installation_id, export_name, &PluginHostOperation::Admit)?,
+            };
             if record.cancelled { return Err(policy::failure(PluginHostErrorCode::Cancelled)); }
             if !matches!(action, Action::Lookup(..)) && (record.expired || record.expires_at_ms <= now_ms()) { return Err(policy::failure(PluginHostErrorCode::ReceiptExpired)); }
             if !tx.query_row("SELECT EXISTS(SELECT 1 FROM typed_tickets t JOIN workspaces w USING(workspace_id) WHERE t.workspace_id=?1 AND t.ticket_id=?2 AND w.state='active')",rusqlite::params![workspace,ticket_id],|row|row.get::<_,bool>(0))? { return Err(policy::failure(PluginHostErrorCode::NotFound)); }

crates/workspace-server/src/server_plugin_data_tests.rs

@@ -106,7 +106,7 @@ fn create_ticket_with_slug(f: &Fixture, body: &str, slug: Option<&str>) -> ticke
 
 // Keep the real installation policy and change only concrete Ticket permissions.
 // A cached resolve or receipt must not bypass these subsequent live changes.
-fn ticket_policy(f: &Fixture, tickets: &[(&str, bool, bool)]) {
+pub(super) fn ticket_policy(f: &Fixture, tickets: &[(&str, bool, bool)]) {
     let store = open_subjektiv_store(&f.api).unwrap();
     let previous = store.subject_permissions(&f.subject).unwrap().unwrap();
     let mut execution: Vec<_> = previous
@@ -166,6 +166,7 @@ async fn host_for_plan(
                 context.grants.connection_slots.insert(slot.clone());
             }
             PluginGrant::TicketRead => context.grants.ticket_read = true,
+            PluginGrant::TicketEvents => (),
             PluginGrant::TicketDataRead {
                 collection,
                 schema_id,

crates/workspace-server/src/server_plugin_diagnostics.rs

@@ -95,19 +95,26 @@ pub(super) async fn diagnostics(
                         .plugins()
                         .iter()
                         .map(|pin| {
-                            subject_authorization::authorize_scope(
-                                &api,
-                                &worker_ref,
-                                &OperationAuthorizationRequest {
-                                    operation: "plugin.admit".into(),
-                                    resource: PermissionResource::PluginInstallation {
-                                        workspace_id: w.clone(),
-                                        installation_id: pin.installation_id().into(),
+                            let scope = |operation: &str| {
+                                subject_authorization::authorize_scope(
+                                    &api,
+                                    &worker_ref,
+                                    &OperationAuthorizationRequest {
+                                        operation: operation.into(),
+                                        resource: PermissionResource::PluginInstallation {
+                                            workspace_id: w.clone(),
+                                            installation_id: pin.installation_id().into(),
+                                        },
                                     },
-                                },
-                            )
-                            .map(|_| ())
-                            .map_err(|error| reason(error.error))
+                                )
+                                .map(|_| ())
+                                .map_err(|error| reason(error.error))
+                            };
+                            scope("plugin.admit")?;
+                            if !pin.projection().services.is_empty() {
+                                scope("plugin.service.consume")?;
+                            }
+                            Ok(())
                         })
                         .collect();
                     store.with_conn_mut(|conn| {
@@ -125,17 +132,25 @@ pub(super) async fn diagnostics(
                                     .map_err(|code| *code)
                                     .and_then(|execution| {
                                         scope?;
-                                        policy::authorize(
-                                            &tx,
-                                            &w,
-                                            execution,
-                                            &saved.plan_id,
-                                            pin.installation_id(),
-                                            &pin.projection().tools,
-                                            &PluginHostOperation::Admit,
-                                        )
-                                        .map(|_| ())
-                                        .map_err(reason)
+                                        if !pin.projection().tools.is_empty() || pin.projection().services.is_empty() {
+                                            policy::authorize(
+                                                &tx,
+                                                &w,
+                                                execution,
+                                                &saved.plan_id,
+                                                pin.installation_id(),
+                                                &pin.projection().tools,
+                                                &PluginHostOperation::Admit,
+                                            ).map_err(reason)?;
+                                        }
+                                        for service in &pin.projection().services {
+                                            policy::authorize_service(
+                                                &tx, &w, execution, &saved.plan_id,
+                                                pin.installation_id(), service,
+                                                &PluginHostOperation::Admit,
+                                            ).map_err(reason)?;
+                                        }
+                                        Ok(())
                                     });
                             let admission = match observed {
                                 Ok(()) => PluginDiagnosticAdmission::Admissible,
@@ -165,6 +180,17 @@ pub(super) async fn diagnostics(
                                     }
                                 })
                                 .collect();
+                            let consumers = tx.prepare(
+                                "SELECT consumer_id FROM plugin_service_consumers WHERE workspace_id=?1 AND plan_id=?2 AND installation_id=?3 ORDER BY service_name LIMIT 32"
+                            )?.query_map(rusqlite::params![w, saved.plan_id, pin.installation_id()], |row| row.get::<_, String>(0))?
+                                .collect::<rusqlite::Result<Vec<_>>>()?;
+                            let service_consumers = consumers.into_iter().map(|id| {
+                                let consumer = crate::plugin_service::load(&tx, &w, &id)?;
+                                Ok(PluginServiceConsumerDiagnostics {
+                                    service_name: consumer.service.clone(),
+                                    diagnostics: crate::plugin_service::diagnostics(&tx, &consumer, chrono::Utc::now().timestamp_millis())?,
+                                })
+                            }).collect::<Result<Vec<_>>>()?;
                             plugins.push(PluginWorkerDiagnostic {
                                 installation_id: pin.installation_id().into(),
                                 package_id: pin.package_id().into(),
@@ -180,9 +206,14 @@ pub(super) async fn diagnostics(
                                         crate::plugin_plan::PluginSurface::Tool => {
                                             PluginDiagnosticSurface::Tool
                                         }
+                                        crate::plugin_plan::PluginSurface::Service => {
+                                            PluginDiagnosticSurface::Service
+                                        }
                                     })
                                     .collect(),
                                 selected_tools: pin.projection().tools.clone(),
+                                selected_services: pin.projection().services.clone(),
+                                service_consumers,
                                 admission,
                                 connections,
                             });

crates/workspace-server/src/server_plugin_host.rs

@@ -7,6 +7,12 @@ use rusqlite::OptionalExtension;
 use server_api::*;
 #[path = "server_plugin_data.rs"]
 mod data;
+#[path = "server_plugin_service.rs"]
+mod service;
+tokio::task_local! { static SERVICE_LEASE: PluginServiceLease; }
+fn service_lease() -> Option<PluginServiceLease> {
+    SERVICE_LEASE.try_with(Clone::clone).ok()
+}
 
 impl WorkspaceApi {
     /// Host plan creation after the canonical reservation and grant/Job binding,
@@ -92,6 +98,29 @@ pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
 fn denied() -> PluginHostError {
     policy::map_error(policy::failure(PluginHostErrorCode::Denied))
 }
+fn validate_log_entry(entry: &PluginLogEntry) -> std::result::Result<(), PluginHostError> {
+    if entry.message.len() > 4096
+        || entry.event_code.is_empty()
+        || entry.event_code.len() > 128
+        || !entry
+            .event_code
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.'))
+        || entry.fields.len() > 32
+        || entry.fields.iter().any(|field| {
+            field.name.len() > 128
+                || field.name.is_empty()
+                || field.name.chars().any(char::is_control)
+                || matches!(&field.value, PluginLogScalar::Text(value) if value.len() > 1024)
+                || matches!(&field.value, PluginLogScalar::Number(value) if !value.is_finite())
+        })
+    {
+        return Err(policy::map_error(policy::failure(
+            PluginHostErrorCode::QuotaExceeded,
+        )));
+    }
+    Ok(())
+}
 fn now_ms() -> i64 {
     chrono::Utc::now().timestamp_millis()
 }
@@ -105,6 +134,20 @@ fn operation_name(op: &PluginHostOperation) -> &'static str {
     }
 }
 
+/// Retain the physical Subject database only for this synchronous check. Each
+/// policy/lineage query still reads its owner; no authorization result or lock
+/// survives the closure, and FeatureStorage continues to enforce lifecycle fences.
+pub(super) fn with_authority_database<T>(
+    api: &WorkspaceApi,
+    check: impl FnOnce() -> crate::Result<T>,
+) -> crate::Result<T> {
+    let _database = api
+        .feature_storage
+        .open(&api.subjektiv_registration)
+        .map_err(|error| Error::Store(error.to_string()))?;
+    check()
+}
+
 async fn check(
     api: Arc<WorkspaceApi>,
     context: ServerRequestContext,
@@ -114,6 +157,25 @@ async fn check(
     consumer: PluginConsumer,
     operation: PluginHostOperation,
 ) -> std::result::Result<policy::AdmittedPluginOperation, PluginHostError> {
+    if let PluginConsumer::Consumer {
+        consumer_id,
+        event_id,
+        export_name,
+    } = &consumer
+    {
+        return service::check_consumer(
+            api,
+            context,
+            workspace,
+            plan_id,
+            installation,
+            consumer_id,
+            event_id,
+            export_name,
+            operation,
+        )
+        .await;
+    }
     let tools = match consumer {
         PluginConsumer::ToolCall {
             tool_call_id,
@@ -172,28 +234,31 @@ async fn check_tools(
     let executor = api.config_store.db_executor();
     executor
         .run(move || {
-            subject_authorization::authorize_scope(
-                &api,
-                &worker,
-                &server_api::authorization::OperationAuthorizationRequest {
-                    operation: operation_name(&operation).into(),
-                    resource: server_api::authorization::PermissionResource::PluginInstallation {
-                        workspace_id: workspace.clone(),
-                        installation_id: installation.clone(),
+            with_authority_database(&api, || {
+                subject_authorization::authorize_scope(
+                    &api,
+                    &worker,
+                    &server_api::authorization::OperationAuthorizationRequest {
+                        operation: operation_name(&operation).into(),
+                        resource:
+                            server_api::authorization::PermissionResource::PluginInstallation {
+                                workspace_id: workspace.clone(),
+                                installation_id: installation.clone(),
+                            },
                     },
-                },
-            )
-            .map_err(|e| e.error)?;
-            api.config_store.with_immediate_transaction(|tx| {
-                policy::authorize(
-                    tx,
-                    &workspace,
-                    &execution,
-                    &plan_id,
-                    &installation,
-                    &tools,
-                    &operation,
                 )
+                .map_err(|e| e.error)?;
+                api.config_store.with_immediate_transaction(|tx| {
+                    policy::authorize(
+                        tx,
+                        &workspace,
+                        &execution,
+                        &plan_id,
+                        &installation,
+                        &tools,
+                        &operation,
+                    )
+                })
             })
         })
         .await
@@ -213,11 +278,12 @@ async fn invocation(
     let store = api.config_store.clone();
     let w = workspace.to_owned();
     let id = invocation_id.to_owned();
-    let record = store
+    let mut record = store
         .db_executor()
         .run(move || store.with_conn(|conn| receipt::lookup(conn, &w, &id)))
         .await
         .map_err(policy::map_error)?;
+    record.service_lease = service_lease();
     let admitted = check(
         api,
         context.clone(),
@@ -394,6 +460,14 @@ async fn execute_with_transport(
     // Human management and a Runtime-only (no Worker) source both fail closed.
     subject_authorization::source_from_context(&context).map_err(|_| denied())?;
     match request {
+        PluginHostRequest::Service { request } => {
+            if service_lease().is_some() {
+                return Err(denied());
+            }
+            service::execute(api, context, workspace, request, transport_override)
+                .await
+                .map(|response| PluginHostResponse::Service { response })
+        }
         request @ (PluginHostRequest::ResolveTicket { .. }
         | PluginHostRequest::ReadTicket { .. }
         | PluginHostRequest::DataGet { .. }
@@ -425,6 +499,9 @@ async fn execute_with_transport(
             installation_id,
             consumer,
         } => {
+            if matches!(consumer, PluginConsumer::Consumer { .. }) {
+                return Err(denied());
+            }
             let admitted = check(
                 api.clone(),
                 context,
@@ -469,6 +546,9 @@ async fn execute_with_transport(
             })
         }
         PluginHostRequest::Cancel { invocation_id } => {
+            if service_lease().is_some() {
+                return Err(denied());
+            }
             // Do not use invocation(): exact cancellation replay must work after
             // the irreversible bit has already been committed.
             let store = api.config_store.clone();
@@ -536,24 +616,7 @@ async fn execute_with_transport(
             .await?;
             // Guest text/fields are untrusted and may contain echoed credentials.
             // Standard diagnostics retain only this safe event, not their payload.
-            if entry.message.len() > 4096
-                || entry.event_code.is_empty()
-                || entry.event_code.len() > 128
-                || !entry
-                    .event_code
-                    .bytes()
-                    .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'.'))
-                || entry.fields.len() > 32 || entry.fields.iter().any(|field| {
-                field.name.len() > 128
-                    || field.name.is_empty()
-                    || field.name.chars().any(char::is_control)
-                    || matches!(&field.value, PluginLogScalar::Text(value) if value.len() > 1024)
-                    || matches!(&field.value, PluginLogScalar::Number(value) if !value.is_finite())
-            }) {
-                return Err(policy::map_error(policy::failure(
-                    PluginHostErrorCode::QuotaExceeded,
-                )));
-            }
+            validate_log_entry(&entry)?;
             tracing::debug!(invocation = %invocation_id, "plugin log event admitted");
             Ok(PluginHostResponse::Logged)
         }

crates/workspace-server/src/server_plugin_host_tests.rs

@@ -4,6 +4,8 @@
 // proof verification, current Subject permissions and all Host checks are real.
 // Run with `cargo test -p yoi-workspace-server --lib plugin_host_tests`.
 // The shared TLS fixture requires openssl; tests never mutate process environment.
+#[path = "server_plugin_authority_storage_tests.rs"]
+mod authority_storage;
 #[path = "server_plugin_data_tests.rs"]
 mod data;
 
@@ -332,7 +334,17 @@ impl Fixture {
     }
 
     fn count(&self, table: &str) -> i64 {
-        assert!(["plugin_invocations", "plugin_effect_receipts", "plugin_data_cells", "plugin_data_receipts", "plugin_data_purges"].contains(&table));
+        assert!(
+            [
+                "plugin_invocations",
+                "plugin_effect_receipts",
+                "plugin_data_cells",
+                "plugin_data_receipts",
+                "plugin_data_purges",
+                "plugin_service_released_proofs",
+            ]
+            .contains(&table)
+        );
         self.api
             .config_store
             .with_conn(|conn| {
@@ -379,6 +391,7 @@ fn set_policy(api: &WorkspaceApi, subject: &str, installation: &str, granted: bo
             "plugin.connection.open",
             "plugin.http.send",
             "plugin.log",
+            "plugin.service.consume",
         ]
         .into_iter()
         .map(|operation| ScopedPermission {
@@ -1350,6 +1363,59 @@ async fn remote_owner_for_with_limit(
     )
 }
 
+fn assert_check_physical_initializations(
+    f: &Fixture,
+    before: crate::feature_storage::OpenObservation,
+    expected: u64,
+) {
+    let observed = f
+        .api
+        .subjektiv_registration
+        .open_observation()
+        .since(before);
+    let directory = f
+        .api
+        .config
+        .database_path
+        .parent()
+        .unwrap()
+        .join("feature-storage")
+        .join(TEST_WORKSPACE_ID)
+        .join("features");
+    let resolved = fs::canonicalize(&directory).unwrap();
+    let ancestors: std::collections::HashSet<_> =
+        directory.ancestors().chain(resolved.ancestors()).collect();
+    assert_eq!(observed.completed_initializations, expected, "{observed:?}");
+    assert_eq!(observed.physical_open_attempts, expected, "{observed:?}");
+    assert_eq!(observed.physical_opens, expected, "{observed:?}");
+    assert_eq!(
+        observed.directory_sync_attempts,
+        expected * ancestors.len() as u64,
+        "{observed:?}"
+    );
+    assert_eq!(
+        observed.directory_sync_successes,
+        observed.directory_sync_attempts
+    );
+}
+
+#[tokio::test]
+async fn tool_check_reuses_physical_subject_database_only_within_each_sync_scope() {
+    let f = Fixture::new(None).await;
+    let owner = f.owner(None).await;
+    for _ in 0..2 {
+        let before = f.api.subjektiv_registration.open_observation();
+        assert!(matches!(
+            owner.call(prepare(&f)).await.unwrap(),
+            PluginHostResponse::Authorized
+        ));
+        // Identity and permission checks are two separate synchronous scopes.
+        // The next call must open again rather than keeping a cross-call cache.
+        assert_check_physical_initializations(&f, before, 2);
+    }
+    f.shutdown().await;
+}
+
 fn prepare(f: &Fixture) -> PluginHostRequest {
     PluginHostRequest::Prepare {
         plan_id: f.plan.plan_id.clone(),

crates/workspace-server/src/server_plugin_runtime_tests.rs

@@ -311,6 +311,8 @@ pub(super) mod plugin_runtime_tests {
                 .plugins
                 .selections
                 .push(manifest::PluginProfileSelection {
+                    services: vec![],
+                    ticket_events: false,
                     ticket_read: false,
                     ticket_data_read: vec![],
                     ticket_data_write: vec![],
@@ -837,9 +839,11 @@ pub(super) mod plugin_runtime_tests {
             .plugins
             .selections
             .push(manifest::PluginProfileSelection {
-                    ticket_read: false,
-                    ticket_data_read: vec![],
-                    ticket_data_write: vec![],
+                services: vec![],
+                ticket_events: false,
+                ticket_read: false,
+                ticket_data_read: vec![],
+                ticket_data_write: vec![],
                 installation_id: f.installed.installation.installation_id.clone(),
                 tools: vec![f.tool.into()],
                 logging: false,
@@ -1188,4 +1192,5 @@ pub(super) mod plugin_runtime_tests {
     async fn pending_tool_call_restores_its_committed_host_identity_before_first_effect() {
         https_through_runtime_and_history(false, true).await;
     }
+    include!("server_plugin_service_runtime_tests.rs");
 }

crates/workspace-server/src/server_plugin_service.rs

@@ -0,0 +1,522 @@
+//! Authenticated Remote and Embedded Service operations share this owner.
+use super::*;
+use crate::plugin_service as store;
+use server_api::authorization::{OperationAuthorizationRequest, PermissionResource};
+
+async fn check_service(
+    api: Arc<WorkspaceApi>,
+    context: ServerRequestContext,
+    workspace: String,
+    plan_id: String,
+    installation: String,
+    name: String,
+    operation: PluginHostOperation,
+    preparing: bool,
+) -> std::result::Result<policy::AdmittedPluginOperation, PluginHostError> {
+    validate_workspace_scope(&api, &workspace).map_err(|e| policy::map_error(e.error))?;
+    let worker = subject_authorization::source_from_context(&context).map_err(|_| denied())?;
+    let execution = if preparing
+        && api
+            .store
+            .has_active_worker_create_reservation(api.workspace_id(), &worker)
+            .map_err(policy::map_error)?
+    {
+        api.plugin_reserved_identity(&worker, None, None)
+            .map_err(policy::map_error)?
+    } else {
+        api.plugin_execution_identity(&worker)
+            .await
+            .map_err(policy::map_error)?
+    };
+    let executor = api.config_store.db_executor();
+    executor
+        .run(move || {
+            with_authority_database(&api, || {
+                for op in ["plugin.service.consume", operation_name(&operation)] {
+                    subject_authorization::authorize_scope(
+                        &api,
+                        &worker,
+                        &OperationAuthorizationRequest {
+                            operation: op.into(),
+                            resource: PermissionResource::PluginInstallation {
+                                workspace_id: workspace.clone(),
+                                installation_id: installation.clone(),
+                            },
+                        },
+                    )
+                    .map_err(|e| e.error)?;
+                }
+                api.config_store.with_immediate_transaction(|tx| {
+                    policy::authorize_service(
+                        tx,
+                        &workspace,
+                        &execution,
+                        &plan_id,
+                        &installation,
+                        &name,
+                        &operation,
+                    )
+                })
+            })
+        })
+        .await
+        .map_err(policy::map_error)
+}
+
+pub(super) async fn check_consumer(
+    api: Arc<WorkspaceApi>,
+    context: ServerRequestContext,
+    workspace: String,
+    plan: String,
+    installation: String,
+    consumer_id: &str,
+    event_id: &str,
+    name: &str,
+    operation: PluginHostOperation,
+) -> std::result::Result<policy::AdmittedPluginOperation, PluginHostError> {
+    let lease = service_lease().ok_or_else(denied)?;
+    if lease.consumer_id != consumer_id {
+        return Err(denied());
+    }
+    let admitted = check_service(
+        api.clone(),
+        context,
+        workspace.clone(),
+        plan.clone(),
+        installation.clone(),
+        name.into(),
+        operation,
+        false,
+    )
+    .await?;
+    let config = api.config_store.clone();
+    let name = name.to_owned();
+    let event = event_id.to_owned();
+    let execution = admitted.plan.plan.execution().clone();
+    config
+        .db_executor()
+        .run(move || {
+            config.with_conn(|conn| {
+                let c = store::fence(conn, &workspace, &lease, now_ms())?;
+                if c.plan_id != plan
+                    || c.installation != installation
+                    || c.service != name
+                    || c.current_event_id.as_deref() != Some(&event)
+                    || c.runtime_id != execution.runtime_id
+                    || c.worker_id != execution.worker_id
+                    || c.pause.is_some()
+                {
+                    return Err(policy::failure(PluginHostErrorCode::Denied));
+                }
+                Ok(())
+            })
+        })
+        .await
+        .map_err(policy::map_error)?;
+    Ok(admitted)
+}
+
+async fn checked_consumer(
+    api: Arc<WorkspaceApi>,
+    context: ServerRequestContext,
+    workspace: String,
+    id: String,
+    lease: Option<PluginServiceLease>,
+) -> std::result::Result<(store::Consumer, policy::AdmittedPluginOperation), PluginHostError> {
+    let config = api.config_store.clone();
+    let w = workspace.clone();
+    let proof = lease.clone();
+    let c = config
+        .db_executor()
+        .run(move || {
+            config.with_conn(|conn| match proof {
+                Some(proof) => store::fence(conn, &w, &proof, now_ms()),
+                None => store::load(conn, &w, &id),
+            })
+        })
+        .await
+        .map_err(policy::map_error)?;
+    let worker = subject_authorization::source_from_context(&context).map_err(|_| denied())?;
+    let checked = check_service(
+        api.clone(),
+        context,
+        workspace,
+        c.plan_id.clone(),
+        c.installation.clone(),
+        c.service.clone(),
+        PluginHostOperation::Admit,
+        false,
+    )
+    .await;
+    let admitted = match checked {
+        Ok(admitted) => admitted,
+        Err(error) => {
+            if c.runtime_id == worker.runtime_id && c.worker_id == worker.worker_id {
+                if let Some(proof) = lease {
+                    let config = api.config_store.clone();
+                    // Do not let an old driver's delayed authorization failure
+                    // pause a newly leased consumer (or mutate diagnostics reads).
+                    let _ = config
+                        .db_executor()
+                        .run(move || {
+                            config.with_immediate_transaction(|tx| {
+                                let current = store::fence(tx, &c.workspace, &proof, now_ms())?;
+                                store::pause(tx, &current, PluginServicePauseReason::Authorization)
+                            })
+                        })
+                        .await;
+                }
+            }
+            return Err(error);
+        }
+    };
+    let execution = admitted.plan.plan.execution();
+    if c.runtime_id != execution.runtime_id || c.worker_id != execution.worker_id {
+        return Err(denied());
+    }
+    Ok((c, admitted))
+}
+
+pub(super) async fn execute(
+    api: Arc<WorkspaceApi>,
+    context: ServerRequestContext,
+    workspace: String,
+    request: PluginServiceRequest,
+    transport_override: Option<Arc<https::HttpsTransport>>,
+) -> std::result::Result<PluginServiceResponse, PluginHostError> {
+    match request {
+        PluginServiceRequest::Prepare {
+            plan_id,
+            installation_id,
+            service_name,
+            operation,
+        } => {
+            let operation = match operation {
+                PluginServiceLifecycleOperation::Admit => PluginHostOperation::Admit,
+                PluginServiceLifecycleOperation::Publish => PluginHostOperation::Publish,
+            };
+            check_service(
+                api,
+                context,
+                workspace,
+                plan_id,
+                installation_id,
+                service_name,
+                operation,
+                true,
+            )
+            .await?;
+            Ok(PluginServiceResponse::Prepared)
+        }
+        PluginServiceRequest::LifecycleLog {
+            plan_id,
+            installation_id,
+            service_name,
+            entry,
+        } => {
+            check_service(
+                api,
+                context,
+                workspace,
+                plan_id.clone(),
+                installation_id,
+                service_name,
+                PluginHostOperation::Log,
+                true,
+            )
+            .await?;
+            validate_log_entry(&entry)?;
+            // Do not retain untrusted guest text/fields, including echoed secrets.
+            tracing::debug!(plan = %plan_id, "plugin service lifecycle log admitted");
+            Ok(PluginServiceResponse::Logged)
+        }
+        PluginServiceRequest::Acquire {
+            plan_id,
+            installation_id,
+            service_name,
+        } => {
+            let admitted = check_service(
+                api.clone(),
+                context,
+                workspace,
+                plan_id,
+                installation_id.clone(),
+                service_name.clone(),
+                PluginHostOperation::Admit,
+                false,
+            )
+            .await?;
+            let declarations = admitted
+                .services
+                .iter()
+                .find(|s| s.name == service_name)
+                .ok_or_else(denied)?;
+            let subscriptions = declarations
+                .subscriptions
+                .iter()
+                .map(|s| match s.as_str() {
+                    "ticket_changed" => Ok(PluginTicketEventType::Changed),
+                    "ticket_deleted" => Ok(PluginTicketEventType::Deleted),
+                    _ => Err(denied()),
+                })
+                .collect::<std::result::Result<Vec<_>, _>>()?;
+            let config = api.config_store.clone();
+            let lease = config
+                .db_executor()
+                .run(move || {
+                    config.with_immediate_transaction(|tx| {
+                        store::acquire(
+                            tx,
+                            &admitted.plan,
+                            &installation_id,
+                            &service_name,
+                            &subscriptions,
+                            now_ms(),
+                        )
+                    })
+                })
+                .await
+                .map_err(policy::map_error)?;
+            Ok(PluginServiceResponse::Acquired { lease })
+        }
+        PluginServiceRequest::Poll { lease } => {
+            let (_, admitted) = checked_consumer(
+                api.clone(),
+                context.clone(),
+                workspace.clone(),
+                lease.consumer_id.clone(),
+                Some(lease.clone()),
+            )
+            .await?;
+            let config = api.config_store.clone();
+            let w = workspace.clone();
+            let proof = lease.clone();
+            let response = config
+                .db_executor()
+                .run(move || {
+                    config.with_immediate_transaction(|tx| {
+                        store::poll(tx, &w, &proof, &admitted.plan, now_ms())
+                    })
+                })
+                .await
+                .map_err(policy::map_error)?;
+            let PluginServiceResponse::Delivery { envelope, .. } = &response else {
+                return Ok(response);
+            };
+            // Subject authorization re-enters the config store for live Worker
+            // lineage. Never call it while holding the consumer write transaction.
+            // The envelope remains Host-local until authorization and a fresh
+            // lease/event fence both succeed; no guest sees an unchecked delivery.
+            let config = api.config_store.clone();
+            let ticket = envelope.ticket_id.clone();
+            let event = envelope.event_id.clone();
+            let kind = envelope.event_type;
+            config
+                .db_executor()
+                .run(move || {
+                    let worker = subject_authorization::source_from_context(&context)
+                        .map_err(|e| e.error)?;
+                    let allowed = subject_authorization::authorize(
+                        &api,
+                        &worker,
+                        &OperationAuthorizationRequest {
+                            operation: "ticket.read".into(),
+                            resource: PermissionResource::Ticket {
+                                workspace_id: workspace.clone(),
+                                ticket_id: ticket.clone(),
+                            },
+                        },
+                    );
+                    config.with_immediate_transaction(|tx| {
+                        let c = store::fence(tx, &workspace, &lease, now_ms())?;
+                        if c.current_event_id.as_deref() != Some(event.as_str()) {
+                            return Err(policy::failure(PluginHostErrorCode::Denied));
+                        }
+                        // Deleted references are metadata only: this does not
+                        // resolve or recreate a Ticket capability.
+                        let exists: bool = tx.query_row(
+                            "SELECT EXISTS(SELECT 1 FROM typed_tickets WHERE workspace_id=?1 AND ticket_id=?2)",
+                            rusqlite::params![workspace, ticket],
+                            |r| r.get(0),
+                        )?;
+                        if allowed.is_err() || (kind == PluginTicketEventType::Changed && !exists) {
+                            store::pause(tx, &c, PluginServicePauseReason::Authorization)?;
+                            return Ok(PluginServiceResponse::Diagnostics {
+                                diagnostics: store::diagnostics(
+                                    tx,
+                                    &store::load(tx, &workspace, &lease.consumer_id)?,
+                                    now_ms(),
+                                )?,
+                            });
+                        }
+                        Ok(response)
+                    })
+                })
+                .await
+                .map_err(policy::map_error)
+        }
+        PluginServiceRequest::Complete {
+            lease,
+            event_id,
+            result,
+        } => {
+            checked_consumer(
+                api.clone(),
+                context,
+                workspace.clone(),
+                lease.consumer_id.clone(),
+                Some(lease.clone()),
+            )
+            .await?;
+            let config = api.config_store.clone();
+            config
+                .db_executor()
+                .run(move || {
+                    config.with_immediate_transaction(|tx| {
+                        store::complete(tx, &workspace, &lease, &event_id, &result, now_ms())
+                    })
+                })
+                .await
+                .map_err(policy::map_error)?;
+            Ok(PluginServiceResponse::Completed)
+        }
+        PluginServiceRequest::Release { lease } => {
+            // Authenticate the same Worker without an active lease fence or
+            // revoked plugin grants: cleanup and lost-response reconciliation
+            // must remain possible after expiry, release, or lease rotation.
+            validate_workspace_scope(&api, &workspace).map_err(|e| policy::map_error(e.error))?;
+            let worker =
+                subject_authorization::source_from_context(&context).map_err(|_| denied())?;
+            let config = api.config_store.clone();
+            config
+                .db_executor()
+                .run(move || {
+                    config.with_immediate_transaction(|tx| {
+                        let c = store::load(tx, &workspace, &lease.consumer_id)?;
+                        if c.runtime_id != worker.runtime_id || c.worker_id != worker.worker_id {
+                            return Err(policy::failure(PluginHostErrorCode::Denied));
+                        }
+                        store::release(tx, &workspace, &lease, now_ms())
+                    })
+                })
+                .await
+                .map_err(policy::map_error)?;
+            Ok(PluginServiceResponse::Released)
+        }
+        PluginServiceRequest::Diagnostics { consumer_id } => {
+            let (c, _) =
+                checked_consumer(api.clone(), context, workspace, consumer_id, None).await?;
+            let config = api.config_store.clone();
+            let diagnostics = config
+                .db_executor()
+                .run(move || config.with_conn(|conn| store::diagnostics(conn, &c, now_ms())))
+                .await
+                .map_err(policy::map_error)?;
+            Ok(PluginServiceResponse::Diagnostics { diagnostics })
+        }
+        PluginServiceRequest::Host { lease, request } => {
+            // The finite invocation DTO cannot represent Service/control nesting.
+            checked_consumer(
+                api.clone(),
+                context.clone(),
+                workspace.clone(),
+                lease.consumer_id.clone(),
+                Some(lease.clone()),
+            )
+            .await?;
+            let response = SERVICE_LEASE
+                .scope(
+                    lease,
+                    Box::pin(execute_with_transport(
+                        api,
+                        context,
+                        workspace,
+                        (*request).into(),
+                        transport_override,
+                    )),
+                )
+                .await?;
+            Ok(PluginServiceResponse::Host {
+                response: Box::new(response.try_into().map_err(|_| denied())?),
+            })
+        }
+    }
+}
+
+fn management_failure(error: Error) -> RepositoryApiError {
+    let error = policy::map_error(error);
+    let (status, code) = match error.code {
+        PluginHostErrorCode::InvalidInput | PluginHostErrorCode::Unsupported => {
+            (400, "plugin_service_invalid")
+        }
+        PluginHostErrorCode::Denied => (403, "plugin_service_denied"),
+        PluginHostErrorCode::NotFound => (404, "plugin_service_not_found"),
+        PluginHostErrorCode::Conflict | PluginHostErrorCode::ReceiptExpired => {
+            (409, "plugin_service_conflict")
+        }
+        // A DB/transport failure is not proof that a recovery mutation failed.
+        _ => (503, "plugin_service_outcome_unknown"),
+    };
+    RepositoryApiError::new(status, code, error.message(), vec![])
+}
+
+impl WorkspaceApi {
+    /// Management observations never run a guest or require a live Runtime.
+    pub async fn plugin_service_diagnostics(
+        &self,
+        workspace: String,
+        actor: RequestActor,
+        consumer_id: String,
+    ) -> std::result::Result<PluginServiceDiagnostics, RepositoryApiError> {
+        super::super::plugin::authorize(self, &workspace, &actor, false).await?;
+        let config = self.config_store.clone();
+        config
+            .db_executor()
+            .run(move || {
+                config.with_conn(|conn| {
+                    let consumer = store::load(conn, &workspace, &consumer_id)?;
+                    store::diagnostics(conn, &consumer, now_ms())
+                })
+            })
+            .await
+            .map_err(management_failure)
+    }
+
+    /// Management adapter for the parent contract dispatcher. This is not a guest
+    /// Host import; the ordinary owner-account verifier precedes the DB owner.
+    pub async fn recover_plugin_service(
+        &self,
+        workspace: String,
+        actor: RequestActor,
+        request: PluginServiceRecoveryRequest,
+    ) -> std::result::Result<PluginServiceDiagnostics, RepositoryApiError> {
+        super::super::plugin::authorize(self, &workspace, &actor, true).await?;
+        let actor = serde_json::to_string(&(&actor.account_id, &actor.user_id))
+            .map_err(|_| management_failure(policy::failure(PluginHostErrorCode::InvalidInput)))?;
+        let config = self.config_store.clone();
+        config
+            .db_executor()
+            .run(move || {
+                config.with_immediate_transaction(|tx| {
+                    store::recover(
+                        tx,
+                        &workspace,
+                        &request.consumer_id,
+                        request.expected_pause,
+                        &request.action,
+                        &request.operation_key,
+                        &actor,
+                        &request.reason,
+                        now_ms(),
+                    )?;
+                    store::diagnostics(
+                        tx,
+                        &store::load(tx, &workspace, &request.consumer_id)?,
+                        now_ms(),
+                    )
+                })
+            })
+            .await
+            .map_err(management_failure)
+    }
+}

crates/workspace-server/src/store.rs

@@ -129,7 +129,7 @@ pub use ticket::{
 };
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 111;
+const LATEST_SCHEMA_VERSION: i64 = 112;
 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";
@@ -492,6 +492,11 @@ const MIGRATIONS: &[Migration] = &[
         name: "Typed Plugin Ticket extension data and atomic mutation receipts",
         apply: migrate_plugin_data_v110_to_v111,
     },
+    Migration {
+        version: 112,
+        name: "Ticket Service outbox and fenced consumers",
+        apply: migrate_plugin_services_v111_to_v112,
+    },
 ];
 
 // Both schema-103 identities shipped. Keep the alternate implementation and accept
@@ -15927,6 +15932,41 @@ fn verify_plugin_data_schema(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
+fn migrate_plugin_services_v111_to_v112(conn: &Connection) -> Result<()> {
+    if current_schema_version(conn)? != 111 {
+        return Err(Error::Store("Service migration requires schema 111".into()));
+    }
+    verify_schema_history(conn, 111)?;
+    let tx = conn.unchecked_transaction()?;
+    ticket::migrate_sqlite_ticket_schema_in_transaction(&tx)
+        .map_err(|e| Error::Store(format!("Ticket outbox migration failed: {e}")))?;
+    tx.execute_batch(crate::plugin_service::SCHEMA_SQL)?;
+    tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(112,'Ticket Service outbox and fenced consumers')",[])?;
+    tx.commit()?;
+    Ok(())
+}
+
+fn verify_plugin_service_schema(conn: &Connection) -> Result<()> {
+    let expected = Connection::open_in_memory()?;
+    expected.execute_batch(crate::plugin_service::SCHEMA_SQL)?;
+    let mut stmt = expected
+        .prepare("SELECT name,sql FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY name")?;
+    for entry in stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))? {
+        let (name, sql) = entry?;
+        let actual: Option<String> = conn
+            .query_row(
+                "SELECT sql FROM sqlite_schema WHERE name=?1",
+                [&name],
+                |r| r.get(0),
+            )
+            .optional()?;
+        if actual.as_deref() != Some(&sql) {
+            return Err(Error::Store("Plugin Service schema has drifted".into()));
+        }
+    }
+    Ok(())
+}
+
 fn migrate_plugin_data_v110_to_v111(conn: &Connection) -> Result<()> {
     if current_schema_version(conn)? != 110 {
         return Err(Error::Store(
@@ -16025,6 +16065,7 @@ fn create_workspace_schema_structure(conn: &Connection) -> Result<()> {
     conn.execute_batch(crate::plugin_plan::LAUNCH_SCHEMA_SQL)?;
     conn.execute_batch(crate::plugin_data::REFERENCE_MIGRATION_SQL)?;
     conn.execute_batch(crate::plugin_data::SCHEMA_SQL)?;
+    conn.execute_batch(crate::plugin_service::SCHEMA_SQL)?;
     Ok(())
 }
 
@@ -17067,6 +17108,7 @@ fn apply_migrations_with_secret_source(
     verify_schema_history(conn, LATEST_SCHEMA_VERSION)?;
     crate::plugin_catalog::verify_with_data(conn)?;
     verify_plugin_data_schema(conn)?;
+    verify_plugin_service_schema(conn)?;
     crate::plugin_plan::verify_plan_schema(conn)?;
     crate::plugin_host::verify_receipt_schema(conn)?;
     crate::plugin_plan::verify_launch_schema(conn)?;
@@ -19356,6 +19398,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
     include!("migration_entrypoint_tests.rs");
     include!("plugin_host_migration_tests.rs");
     include!("plugin_data_migration_tests.rs");
+    include!("plugin_service_migration_tests.rs");
     include!("legacy_evidence_column_tests.rs");
 
     #[test]
@@ -19774,6 +19817,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                 WorkspaceSchemaMigrationStep {version:109, name:"Durable Plugin Host invocations and HTTPS effect receipts".into()},
                 WorkspaceSchemaMigrationStep {version:110, name:"Worker Plugin launch selection associations".into()},
                 WorkspaceSchemaMigrationStep {version:111, name:"Typed Plugin Ticket extension data and atomic mutation receipts".into()},
+                WorkspaceSchemaMigrationStep {version:112, name:"Ticket Service outbox and fenced consumers".into()},
             ]
         );
 
@@ -19895,6 +19939,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
 (109, "Durable Plugin Host invocations and HTTPS effect receipts".to_string()),
 (110, "Worker Plugin launch selection associations".to_string()),
 (111, "Typed Plugin Ticket extension data and atomic mutation receipts".to_string()),
+(112, "Ticket Service outbox and fenced consumers".to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -23187,7 +23232,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
                          (108, "Subject Ticket responsibility and current-body notification routing".into()),
                          (109, "Durable Plugin Host invocations and HTTPS effect receipts".into()),
                          (110, "Worker Plugin launch selection associations".into()),
-                         (111, "Typed Plugin Ticket extension data and atomic mutation receipts".into())]);
+                         (111, "Typed Plugin Ticket extension data and atomic mutation receipts".into()),
+                         (112, "Ticket Service outbox and fenced consumers".into())]);
                 for (table, rows) in unchanged.iter().zip(before.iter()) {
                     assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}");
                 }

crates/workspace-server/src/subject_assignment_migration.rs

@@ -32,7 +32,11 @@ pub(super) fn install(conn: &Connection) -> Result<()> {
             "Subject assignment migration requires a caller-owned transaction".into(),
         ));
     }
-    ticket::migrate_sqlite_ticket_schema_in_transaction(conn).map_err(|error| {
+    ticket::migrate_historical_sqlite_ticket_schema_in_transaction(
+        conn,
+        ticket::HistoricalTicketSchemaTarget::Subjects14,
+    )
+    .map_err(|error| {
         Error::Store(format!(
             "Ticket Subject component composition failed: {error}"
         ))
@@ -58,7 +62,11 @@ pub(super) fn install(conn: &Connection) -> Result<()> {
 /// Ticket. Workspace verifies only its two complete trigger definitions/targets,
 /// followed by host database integrity and FK validation. No repair is performed.
 pub(super) fn verify(conn: &Connection) -> Result<()> {
-    ticket::verify_sqlite_ticket_schema(conn).map_err(|error| {
+    ticket::verify_historical_sqlite_ticket_schema(
+        conn,
+        ticket::HistoricalTicketSchemaTarget::Subjects14,
+    )
+    .map_err(|error| {
         Error::Store(format!(
             "Ticket Subject schema verification failed: {error}"
         ))

crates/workspace-server/src/subject_assignment_migration_tests.rs

@@ -770,11 +770,11 @@ mod subject_assignment_migration_tests {
         let old_components = component_history(&conn);
         {
             let tx = conn.unchecked_transaction().unwrap();
-            ticket::migrate_sqlite_ticket_schema_in_transaction(&tx).unwrap();
+            ticket::migrate_historical_sqlite_ticket_schema_in_transaction(&tx, ticket::HistoricalTicketSchemaTarget::Subjects14).unwrap();
             tx.commit().unwrap();
         }
         assert_component14_append(&conn, &old_components);
-        ticket::verify_sqlite_ticket_schema(&conn).unwrap();
+        ticket::verify_historical_sqlite_ticket_schema(&conn, ticket::HistoricalTicketSchemaTarget::Subjects14).unwrap();
         // The host must not append the owning diagnostic again or issue even a
         // redundant diagnostic UPDATE for any already-diagnosed legacy row.
         conn.execute_batch(
@@ -1070,7 +1070,14 @@ mod subject_assignment_migration_tests {
                 expected
             );
             assert_eq!(history.len(), host.len() + expected.len());
-            assert_component14_append(conn, component);
+            let history = component_history(conn);
+            assert_eq!(&history[..component.len()], component, "owner receipt prefix is preserved");
+            assert_eq!(history.len(), component.len() + 2);
+            let tail: Vec<(i64, String, String)> = conn.prepare("SELECT version,name,applied_at FROM ticket_schema_migrations WHERE version>=14 ORDER BY version").unwrap()
+                .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?))).unwrap().collect::<rusqlite::Result<_>>().unwrap();
+            assert_eq!(tail.iter().map(|(v,name,_)| (*v,name.as_str())).collect::<Vec<_>>(), [(14,"Subject Ticket responsibility"), (15,"Ticket Service transactional outbox")]);
+            assert!(tail.iter().all(|(_,_,at)| !at.is_empty()));
+            ticket::verify_sqlite_ticket_schema(conn).unwrap();
             crate::plugin_catalog::verify_with_data(conn).unwrap();
             plugin_plan::verify_plan_schema(conn).unwrap();
             verify(conn).unwrap();
@@ -1134,6 +1141,7 @@ mod subject_assignment_migration_tests {
                 111,
                 "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
             ));
+            tail.push((112, "Ticket Service outbox and fenced consumers".into()));
             tail
         }
 
@@ -1183,7 +1191,7 @@ mod subject_assignment_migration_tests {
         fn plugin_schema(conn: &Connection) -> Rows {
             fixture_query_rows(
                 conn,
-                "SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema WHERE tbl_name LIKE 'plugin_%' AND tbl_name NOT LIKE 'plugin_data_%' AND tbl_name NOT IN ('plugin_invocations','plugin_effect_receipts','plugin_references') ORDER BY type,name",
+                "SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema WHERE tbl_name LIKE 'plugin_%' AND tbl_name NOT LIKE 'plugin_data_%' AND tbl_name NOT LIKE 'plugin_service_%' AND tbl_name NOT IN ('plugin_invocations','plugin_effect_receipts','plugin_references') ORDER BY type,name",
             )
         }
 

crates/workspace-server/src/ticket_post_upgrade_tests.rs

@@ -83,7 +83,15 @@ fn ticket_posts_schema93_preserves_initial_body_creator_comments_and_archived_au
         assert_eq!(stored_fixture_rows(conn,"typed_ticket_event_references"),old_references);
         let markers = stored_fixture_rows(conn,"ticket_schema_migrations");
         assert_eq!(&markers[..old_markers.len()],old_markers.as_slice());
-        assert_eq!(markers.len(),old_markers.len()+3);
+        assert_eq!(markers.len(),old_markers.len()+4);
+        let new_markers: Vec<(i64,String)> = conn.prepare("SELECT version,name FROM ticket_schema_migrations WHERE version>11 ORDER BY version")?.query_map([],|r|Ok((r.get(0)?,r.get(1)?)))?.collect::<rusqlite::Result<_>>()?;
+        assert_eq!(new_markers,vec![
+            (12,"Canonical Ticket posts".into()),
+            (13,"Ticket content change facts".into()),
+            (14,"Subject Ticket responsibility".into()),
+            (15,"Ticket Service transactional outbox".into()),
+        ]);
+        assert_eq!(conn.query_row("SELECT COUNT(*) FROM typed_ticket_service_outbox",[],|r|r.get::<_,i64>(0))?,0,"migration does not invent Service delivery facts");
         assert_eq!(conn.query_row("SELECT COUNT(*) FROM typed_ticket_content_changes",[],|r|r.get::<_,i64>(0))?,0,"migration does not invent content change facts");
         assert_eq!(conn.query_row("SELECT COUNT(*) FROM ticket_assignment_content_inputs",[],|r|r.get::<_,i64>(0))?,0,"migration does not turn historical edits into new inputs");
         let initial: (String,Option<String>,Option<String>,String,String) = conn.query_row("SELECT e.body,t.created_by,e.author,e.kind,e.event_ref FROM typed_tickets t JOIN typed_ticket_events e USING(workspace_id,ticket_id) WHERE t.ticket_id='authored' AND e.event_index=-1",[],|row|Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?,row.get(4)?)))?;
@@ -140,6 +148,19 @@ fn ticket_posts_schema93_preserves_initial_body_creator_comments_and_archived_au
                 )?,
                 1
             );
+            assert_eq!(
+                conn.query_row(
+                    "SELECT name FROM ticket_schema_migrations WHERE version=15",
+                    [],
+                    |r| r.get::<_, String>(0)
+                )?,
+                "Ticket Service transactional outbox"
+            );
+            assert_eq!(
+                conn.query_row("SELECT COUNT(*) FROM typed_ticket_service_outbox", [], |r| r
+                    .get::<_, i64>(0))?,
+                0
+            );
             Ok(())
         })
         .unwrap();

crates/workspace-server/src/ticket_tag_upgrade_tests.rs

@@ -266,7 +266,7 @@ fn assert_retained_classification_upgrade(workspace_version: i64, component: i64
     let store = SqliteWorkspaceStore::open_for_test(&path).unwrap();
     store.with_conn(|conn| {
         assert_eq!(current_schema_version(conn)?,LATEST_SCHEMA_VERSION);
-        assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,14);
+        assert_eq!(conn.query_row("SELECT MAX(version) FROM ticket_schema_migrations",[],|row|row.get::<_,i64>(0))?,15);
         assert_eq!(fixture_auth_rows(conn),auth_before);
         for (table, rows) in preserved_tables.iter().zip(&preserved) {
             let mut migrated = stored_fixture_rows(conn, table);
@@ -500,7 +500,7 @@ fn workspace90_cases9_add_classification_without_reinterpreting_reasoned_history
                     [],
                     |row| row.get::<_, i64>(0)
                 )?,
-                14
+                15
             );
             assert_eq!(
                 stored_fixture_rows(conn, "typed_ticket_case_history"),

crates/workspace-server/src/workspace_deletion.rs

@@ -16,6 +16,13 @@ use crate::{Error, Result};
 /// Explicit domain-owned purge inventory. The deletion operation tombstone is intentionally
 /// excluded so retries and audit remain available after the Workspace row is gone.
 const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
+    "plugin_service_recovery_audit",
+    // Consumer deletion cascades its released proofs. Their retention guard stays
+    // installed: deleting proofs directly while the consumer exists is forbidden.
+    "plugin_service_consumers",
+    "plugin_service_released_proofs",
+    "plugin_service_retention",
+    "typed_ticket_service_outbox",
     "plugin_data_receipts",
     "plugin_data_purges",
     "plugin_data_holders",
@@ -507,16 +514,17 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
             }
             let workspace_id = operation.response.workspace_id.clone();
 
-            // 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.
+            // Service recovery audit, 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 ticket_delete_guards = Vec::new();
             for (table, trigger) in [
+                ("plugin_service_recovery_audit", "plugin_service_audit_retained"),
                 ("plugin_data_receipts", "plugin_data_receipt_retained"),
                 ("plugin_data_purges", "plugin_data_purge_retained"),
                 ("plugin_effect_receipts", "plugin_effect_retained"),
@@ -1305,6 +1313,16 @@ mod tests {
 
     #[test]
     fn workspace_purge_removes_plugin_references_and_catalog_without_cross_workspace_loss() {
+        let position = |table| {
+            WORKSPACE_DELETION_PURGE_TABLES
+                .iter()
+                .position(|name| *name == table)
+                .unwrap()
+        };
+        assert!(position("plugin_service_recovery_audit") < position("plugin_service_consumers"));
+        assert!(position("plugin_service_consumers") < position("plugin_service_released_proofs"));
+        assert!(position("plugin_service_consumers") < position("plugin_execution_plans"));
+        assert!(position("plugin_service_consumers") < position("plugin_installations"));
         let (store, owner, workspace_id) = setup();
         for workspace in ["workspace-a", "workspace-b"] {
             store.with_conn(|conn| {
@@ -1319,6 +1337,11 @@ mod tests {
                 conn.execute("INSERT INTO plugin_execution_plans VALUES(?1,'plan','actor','create',?2,'{}','active')", params![workspace,digest])?;
                 conn.execute("INSERT INTO plugin_invocations(workspace_id,invocation_id,plan_id,plan_fingerprint,installation_id,runtime_id,worker_id,execution_json,consumer_key,consumer_json,created_at_ms,expires_at_ms) VALUES(?1,'invocation','plan',?2,'installation','runtime','worker','{}','[\"tool_call\",\"call\"]','{}',1,1000)", params![workspace,digest])?;
                 conn.execute("INSERT INTO plugin_effect_receipts(workspace_id,invocation_id,effect_key,fingerprint,admitted_at_ms,state,error_json) VALUES(?1,'invocation','key',?2,2,'pending',?3)", params![workspace,digest,r#"{"code":"indeterminate","effect":"indeterminate","diagnostic_id":"safe"}"#])?;
+                conn.execute("INSERT INTO plugin_service_consumers(workspace_id,consumer_id,runtime_id,worker_id,installation_id,plan_id,service_name,subscriptions_json,cursor) VALUES(?1,'consumer','runtime','worker','installation','plan','service','[\"changed\",\"deleted\"]',1)", [workspace])?;
+                conn.execute("INSERT INTO plugin_service_released_proofs(workspace_id,consumer_id,lease_token,released_at_ms) VALUES(?1,'consumer','released-token',2)", [workspace])?;
+                conn.execute("INSERT INTO plugin_service_recovery_audit(workspace_id,operation_key,consumer_id,fingerprint,actor,reason,action_json,at_ms) VALUES(?1,'recovery','consumer','fingerprint','owner','Explicit recovery','{}',3)", [workspace])?;
+                conn.execute("INSERT INTO plugin_service_retention(workspace_id,removed_through) VALUES(?1,1)", [workspace])?;
+                conn.execute("INSERT INTO typed_ticket_service_outbox(event_id,workspace_id,ticket_id,event_type,occurred_at,domain_event_ref,content_digest,correlation_id,causation_id) VALUES(?1,?2,'deleted-ticket','deleted','2026-01-01T00:00:00Z','deleted-ticket:0',?3,'correlation','causation')", params![if workspace == "workspace-a" { "a".repeat(32) } else { "b".repeat(32) },workspace,"c".repeat(64)])?;
                 for table in ["plugin_effect_receipts", "plugin_invocations", "plugin_execution_plans"] {
                     assert!(conn.execute(&format!("DELETE FROM {table} WHERE workspace_id=?1"), [workspace]).is_err());
                 }
@@ -1329,6 +1352,11 @@ mod tests {
             store
                 .with_conn(|conn| {
                     [
+                        "plugin_service_recovery_audit",
+                        "plugin_service_consumers",
+                        "plugin_service_released_proofs",
+                        "plugin_service_retention",
+                        "typed_ticket_service_outbox",
                         "plugin_effect_receipts",
                         "plugin_invocations",
                         "plugin_execution_plans",
@@ -1350,8 +1378,93 @@ mod tests {
                 })
                 .unwrap()
         };
+        let service_rows = |workspace: &str| {
+            store
+                .with_conn(|conn| {
+                    [
+                        "plugin_service_recovery_audit",
+                        "plugin_service_consumers",
+                        "plugin_service_released_proofs",
+                        "plugin_service_retention",
+                        "typed_ticket_service_outbox",
+                    ]
+                    .map(|table| {
+                        let mut statement = conn.prepare(&format!(
+                            "SELECT * FROM {table} WHERE workspace_id=?1 ORDER BY 1"
+                        ))?;
+                        let columns = statement.column_count();
+                        let rows = statement
+                            .query_map([workspace], |row| {
+                                (0..columns)
+                                    .map(|index| row.get::<_, rusqlite::types::Value>(index))
+                                    .collect::<rusqlite::Result<Vec<_>>>()
+                            })?
+                            .collect::<rusqlite::Result<Vec<_>>>()?;
+                        Ok(rows)
+                    })
+                    .into_iter()
+                    .collect::<Result<Vec<_>>>()
+                })
+                .unwrap()
+        };
+        let service_guards = || {
+            store
+                .with_conn(|conn| {
+                    let mut statement = conn.prepare(
+                        "SELECT name,sql FROM sqlite_schema WHERE type='trigger'
+                         AND tbl_name IN ('plugin_service_consumers',
+                             'plugin_service_released_proofs','plugin_service_recovery_audit')
+                         ORDER BY name",
+                    )?;
+                    let guards = statement
+                        .query_map([], |row| {
+                            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
+                        })?
+                        .collect::<rusqlite::Result<Vec<_>>>()?;
+                    Ok(guards)
+                })
+                .unwrap()
+        };
+        let assert_service_retained = |workspace: &str| {
+            store
+                .with_conn(|conn| {
+                    for (table, column) in [
+                        ("plugin_service_recovery_audit", "actor"),
+                        ("plugin_service_released_proofs", "released_at_ms"),
+                    ] {
+                        assert!(
+                            conn.execute(
+                                &format!("DELETE FROM {table} WHERE workspace_id=?1"),
+                                [workspace],
+                            )
+                            .is_err(),
+                            "ordinary deletion must retain {table}"
+                        );
+                        assert!(
+                            conn.execute(
+                                &format!(
+                                    "UPDATE {table} SET {column}={column} WHERE workspace_id=?1"
+                                ),
+                                [workspace],
+                            )
+                            .is_err(),
+                            "ordinary updates must not rewrite {table}"
+                        );
+                    }
+                    Ok(())
+                })
+                .unwrap();
+        };
+        let guards_before = service_guards();
+        let target_service_before = service_rows("workspace-a");
+        let other_service_before = service_rows("workspace-b");
+        assert_service_retained("workspace-a");
+        assert_service_retained("workspace-b");
         let other = counts("workspace-b");
-        assert_eq!(counts("workspace-a"), vec![1, 1, 1, 4, 1, 1, 1, 1]);
+        assert_eq!(
+            counts("workspace-a"),
+            vec![1, 1, 1, 1, 1, 1, 1, 1, 4, 1, 1, 1, 1]
+        );
         let preflight = store
             .workspace_deletion_preflight(&owner, &workspace_id)
             .unwrap();
@@ -1371,13 +1484,23 @@ mod tests {
             conn.execute_batch("CREATE TRIGGER plugin_purge_failure BEFORE DELETE ON workspaces WHEN OLD.workspace_id='workspace-a' BEGIN SELECT RAISE(ABORT,'injected purge failure'); END;")?;
             Ok(())
         }).unwrap();
+        let error = store
+            .finalize_workspace_deletion(&operation.operation_id)
+            .unwrap_err();
         assert!(
-            store
-                .finalize_workspace_deletion(&operation.operation_id)
-                .is_err()
+            error.to_string().contains("injected purge failure"),
+            "{error}"
+        );
+        assert_eq!(
+            counts("workspace-a"),
+            vec![1, 1, 1, 1, 1, 1, 1, 1, 4, 1, 1, 1, 1]
         );
-        assert_eq!(counts("workspace-a"), vec![1, 1, 1, 4, 1, 1, 1, 1]);
         assert_eq!(counts("workspace-b"), other);
+        assert_eq!(service_rows("workspace-a"), target_service_before);
+        assert_eq!(service_rows("workspace-b"), other_service_before);
+        assert_eq!(service_guards(), guards_before);
+        assert_service_retained("workspace-a");
+        assert_service_retained("workspace-b");
         store
             .with_conn(|conn| {
                 crate::plugin_host::verify_receipt_schema(conn)?;
@@ -1385,15 +1508,21 @@ mod tests {
                 Ok(())
             })
             .unwrap();
+        let completed = store
+            .finalize_workspace_deletion(&operation.operation_id)
+            .unwrap();
+        assert_eq!(completed.state, WorkspaceDeletionState::Succeeded);
         assert_eq!(
             store
                 .finalize_workspace_deletion(&operation.operation_id)
-                .unwrap()
-                .state,
-            WorkspaceDeletionState::Succeeded
+                .unwrap(),
+            completed
         );
-        assert_eq!(counts("workspace-a"), vec![0; 8]);
+        assert_eq!(counts("workspace-a"), vec![0; 13]);
         assert_eq!(counts("workspace-b"), other);
+        assert_eq!(service_rows("workspace-b"), other_service_before);
+        assert_eq!(service_guards(), guards_before);
+        assert_service_retained("workspace-b");
         store
             .with_conn(|conn| {
                 crate::plugin_host::verify_receipt_schema(conn)?;

crates/workspace-server/tests/fixtures/README.md

@@ -0,0 +1,21 @@
+# Service Component fixture
+
+`ticket-service-v2.component.wasm` is the authored Rust/PDK guest from
+`crates/plugin-runtime/tests/service-guest`, not a synthetic Component.
+Its manifest remains the source manifest there (Service world 2.2, Host imports
+2.0/2.1). The workspace-server integration suite installs and validates the
+package, fetches the pinned artifact, and executes the Component through the
+production Feature Service task/owner adapters.
+
+Regenerate with wasm32-unknown-unknown std and `lld` available:
+
+```sh
+cargo test -p yoi-plugin-runtime --test service_component
+cp target/plugin-service-fixture/service.component.wasm \
+  crates/workspace-server/tests/fixtures/ticket-service-v2.component.wasm
+```
+
+The first command builds the guest source using `--locked --offline --release`,
+then uses `wit_component::ComponentEncoder` to validate/encode it. No E2E feature
+flag, external service, or process-global environment mutation is needed for the
+workspace-server suite; its only HTTPS peer is the isolated shared TLS fixture.