feat: connect Subject Ticket assignments across public execution paths

crates/ticket/README.md

@@ -49,3 +49,38 @@ record (`changes` includes `body`), never a later title-only edit. Missing or em
 creator identities remain explicitly nullable, not inferred from later editors.
 The retained base SQL and frozen fixture schemas remain unchanged; fresh creation
 composes the ordered post migration.
+
+## Subject responsibility schema
+
+Component migration 14 owns the Subject principal CHECKs and assignment/current,
+operation, and legacy tracking/history guards. Fresh standalone stores and Workspace
+hosts use the same `migrate_sqlite_ticket_schema_in_transaction(&Connection)` startup
+composition; the host owns its surrounding transaction, authentication, live Subject
+body bindings, delivery/start fences, and notification routing. Request paths only
+verify the owning schema and never install a host-dependent fixture or fallback.
+
+Retained host startup migrations must instead call
+`migrate_historical_sqlite_ticket_schema_in_transaction(&Connection, HistoricalTicketSchemaTarget)`
+with the explicit `Cases10`, `Tracking11`, `Posts12`, or `Content13` boundary. This
+bounded API shares the latest composer's migration engine; it cannot select arbitrary
+versions or downgrade a current store. If the applied version is already newer, it
+verifies that actual contract without replaying old steps or relabelling receipts.
+Original baseline receipts are retained, including v6–9 baselines; their ordered
+v7/10 composition receipts are appended instead of overwriting the baseline. Previously
+retained normalized baseline10 histories remain supported. Neither composer is a
+request-path API or a grant of legacy execution authority.
+
+The additive migration leaves applied core/tracking SQL unchanged. It preflights
+complete known table generations (standalone or Workspace-parent FKs), widens only
+the three principal CHECKs with exact-SQL CAS, restores `writable_schema` on failure,
+and checks integrity before accepting the migration. This avoids rebuilding immutable
+parents and their host-owned FK graph inside a host transaction. Unsupported/mixed
+tables and weakened guards fail closed; a trigger name alone never selects a schema
+generation. Migration history selects the exact verified contract.
+
+Old `workspace_agent` relationships, handling/results, audit source JSON, and receipts
+remain readable as historical data, with a nonoperational diagnostic. They are not
+converted to an arbitrary Subject or replayed as new input. Newly authorized Subject
+Assign/reissue creates a distinct identity. Tracking sources use the Rust/JSON field
+`subject_id`; historical `agent_key` audit JSON remains an unmodified JSON value,
+not an alias accepted as current Subject authority.

crates/ticket/src/assignment_tracking.rs

@@ -8,7 +8,7 @@
 //! Close/reopen does not change handling; current responsibility can report after close
 //! without restoring execution grants. Retained unknown handling is not inferred waiting/done.
 //!
-//! The host authenticates the current workspace-agent Worker and its actual Runtime
+//! The host authenticates the current Subject body Worker and its actual Runtime
 //! binding, including retries, inside the same write transaction as receipt comparison,
 //! current-assignment/event CAS, audit, and result. Never split that transaction around
 //! awaits. A replayed receipt is not a fresh event observation: reread before a new update,
@@ -48,7 +48,7 @@ pub struct TicketAssignmentTrackingUpdate {
     pub tracking: TicketAssignmentTracking,
 }
 
-/// Backend-owned authenticated context, never decoded from model input. Agent
+/// Backend-owned authenticated context, never decoded from model input. Subject
 /// principal remains stable across Worker exchange; exact replay binds to the
 /// authenticated concrete sender and its binding, not merely a claimed role.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -56,7 +56,7 @@ pub struct TicketAssignmentTrackingSourceContext {
     pub actor: String,
     pub runtime_id: String,
     pub worker_id: String,
-    pub agent_key: String,
+    pub subject_id: String,
     pub authentication_binding_id: String,
 }
 
@@ -87,6 +87,9 @@ pub struct TicketAssignmentTrackingProjection {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum TicketAssignmentPrincipalSnapshot {
+    Subject {
+        subject_id: String,
+    },
     User {
         account_id: String,
     },
@@ -156,11 +159,11 @@ fn validate_update(
     ]
     .into_iter()
     .any(|value| !valid_identity(value))
-        || source.agent_key != "workspace-orchestrator"
+        || !valid_identity(&source.subject_id)
+        || source.subject_id == "workspace-orchestrator"
     {
         return Err(TicketError::Conflict(
-            "tracking requires bounded identities and authenticated workspace-orchestrator source"
-                .into(),
+            "tracking requires bounded identities and authenticated Subject source".into(),
         ));
     }
     for text in [
@@ -216,7 +219,8 @@ pub fn sqlite_ticket_assignment_tracking(
         .transpose()?;
     // Legacy migration may retain current responsibility without evidence of
     // handling status. Its real Assign audit is still an immutable observation
-    // against which the first explicit self-report can CAS, even after close.
+    // for a later explicitly authorized operation, even after close. This does
+    // not authorize a legacy agent relationship to report as a Subject.
     // Never manufacture a tracking event/status or repair a missing audit. Once
     // tracking exists, only its event identity is a valid observation.
     let last_event_id = if current && tracking.is_none() && last_event_id.is_none() {
@@ -238,6 +242,9 @@ pub fn sqlite_ticket_assignment_tracking(
     let missing =
         || TicketError::Conflict("stored assignment principal identity is incomplete".into());
     let principal = match kind.as_str() {
+        "subject" => TicketAssignmentPrincipalSnapshot::Subject {
+            subject_id: principal_id.ok_or_else(missing)?,
+        },
         "workspace_agent" => TicketAssignmentPrincipalSnapshot::WorkspaceAgent {
             agent_key: principal_id.ok_or_else(missing)?,
         },
@@ -441,10 +448,10 @@ pub fn sqlite_update_ticket_assignment_tracking_in_transaction(
             "tracking assignment is removed/replaced or last event changed".into(),
         ));
     }
-    let principal_matches:bool=conn.query_row("SELECT principal_kind='workspace_agent' AND principal_id=?3 FROM ticket_worker_assignments WHERE workspace_id=?1 AND assignment_id=?2",params![workspace_id,update.assignment_id,source.agent_key],|r|r.get(0)).map_err(sqlite_err)?;
+    let principal_matches:bool=conn.query_row("SELECT principal_kind='subject' AND principal_id=?3 FROM ticket_worker_assignments WHERE workspace_id=?1 AND assignment_id=?2",params![workspace_id,update.assignment_id,source.subject_id],|r|r.get(0)).map_err(sqlite_err)?;
     if !principal_matches {
         return Err(TicketError::Conflict(
-            "authenticated agent does not own this assignment".into(),
+            "authenticated Subject does not own this assignment".into(),
         ));
     }
     let event_id=format!("tracking-{}",allocate_record_id(unix_epoch_millis_now().unwrap_or(0), |candidate| conn.query_row("SELECT EXISTS(SELECT 1 FROM typed_ticket_assignment_tracking_history WHERE workspace_id=?1 AND event_id=?2)",params![workspace_id,format!("tracking-{candidate}")],|r|r.get::<_,bool>(0)).unwrap_or(true)).map_err(sqlite_err)?);
@@ -572,7 +579,10 @@ fn reconcile_legacy_tracking(conn: &Connection) -> Result<()> {
     Ok(())
 }
 
-pub(crate) fn verify_assignment_tracking_schema(conn: &Connection) -> Result<()> {
+pub(crate) fn verify_assignment_tracking_schema(
+    conn: &Connection,
+    subject_contract: bool,
+) -> Result<()> {
     // Object names alone would accept same-named no-op triggers after drift.
     let canonical = Connection::open_in_memory().map_err(sqlite_err)?;
     canonical
@@ -603,6 +613,13 @@ pub(crate) fn verify_assignment_tracking_schema(conn: &Connection) -> Result<()>
         }
         result.replace("ifnotexists", "")
     };
+    // Generation is selected by the owning migration ledger, never by a trigger name.
+    if subject_contract {
+        canonical
+            .execute_batch(crate::subject_assignment::SUBJECT_GUARDS)
+            .map_err(sqlite_err)?;
+        crate::subject_assignment::verify(conn)?;
+    }
     for name in [
         "typed_ticket_assignment_tracking_history",
         "typed_ticket_assignment_tracking_operations",
@@ -811,6 +828,14 @@ pub(crate) fn verify_assignment_tracking_schema(conn: &Connection) -> Result<()>
                 |r| r.get(0),
             )
             .map_err(sqlite_err)?;
+        let expected_sql = if subject_contract {
+            expected_sql.replace(
+                "CHECK(principal_kind IN ('user', 'worker', 'workspace_agent'))",
+                "CHECK(principal_kind IN ('user', 'worker', 'workspace_agent', 'subject'))",
+            )
+        } else {
+            expected_sql
+        };
         let expected_sql = normalize(expected_sql);
         let actual_sql = normalize(actual_sql);
         let mut remainder = expected_sql.as_str();

crates/ticket/src/assignment_tracking_tests.rs

@@ -25,11 +25,20 @@ fn seed_ticket(conn: &Connection, workspace: &str, ticket: &str) {
 }
 
 fn assign(conn: &Connection, id: &str, event: &str) {
+    assign_principal(conn, id, event, "subject", "subject-a");
+}
+
+// Historical schema fixture only: never infer an authority fallback from DB shape.
+fn assign_legacy(conn: &Connection, id: &str, event: &str) {
+    assign_principal(conn, id, event, "workspace_agent", "workspace-orchestrator");
+}
+
+fn assign_principal(conn: &Connection, id: &str, event: &str, kind: &str, principal: &str) {
     let previous: Option<String> = conn.query_row("SELECT assignment_id FROM ticket_current_worker_assignments WHERE workspace_id='w' AND ticket_id='t' AND role='orchestrator'",[],|r|r.get(0)).optional().unwrap();
     conn.execute_batch("BEGIN IMMEDIATE").unwrap();
-    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t',?1,'orchestrator','workspace_agent','workspace-orchestrator','author','2026-10-10T00:00:00Z')",[id]).unwrap();
+    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t',?1,'orchestrator',?2,?3,'author','2026-10-10T00:00:00Z')",params![id,kind,principal]).unwrap();
     conn.execute("DELETE FROM ticket_current_worker_assignments WHERE workspace_id='w' AND ticket_id='t' AND role='orchestrator'",[]).unwrap();
-    conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES ('w','t',?1,'orchestrator','workspace_agent','workspace-orchestrator','now')",[id]).unwrap();
+    conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES ('w','t',?1,'orchestrator',?2,?3,'now')",params![id,kind,principal]).unwrap();
     conn.execute("INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,previous_assignment_id,actor,created_at,operation_id) VALUES ('w','t','orchestrator',?2,?4,?1,?3,'author','now',?2)",params![id,event,previous,if previous.is_some(){"reassigned"}else{"assigned"}]).unwrap();
     conn.execute_batch("COMMIT").unwrap();
 }
@@ -39,7 +48,7 @@ fn source() -> TicketAssignmentTrackingSourceContext {
         actor: "orchestrator".into(),
         runtime_id: "runtime".into(),
         worker_id: "worker".into(),
-        agent_key: "workspace-orchestrator".into(),
+        subject_id: "subject-a".into(),
         authentication_binding_id: "binding".into(),
     }
 }
@@ -92,7 +101,7 @@ fn assignment_initializes_waiting_and_audit_atomically() {
     assert_eq!(history.len(), 1);
     assert_eq!(history[0].event_id, "assigned-a");
     conn.execute_batch("BEGIN IMMEDIATE").unwrap();
-    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t','rolled-back','orchestrator','workspace_agent','workspace-orchestrator','author','now')",[]).unwrap();
+    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES ('w','t','rolled-back','orchestrator','subject','subject-a','author','now')",[]).unwrap();
     conn.execute_batch("ROLLBACK").unwrap();
     assert!(
         sqlite_ticket_assignment_tracking(&conn, "w", "t", "rolled-back")
@@ -222,13 +231,13 @@ fn tracking_cas_and_receipts_bind_full_payload_and_authenticated_source() {
             1 => sender.runtime_id = "other".into(),
             2 => sender.worker_id = "replacement".into(),
             3 => sender.authentication_binding_id = "replacement-binding".into(),
-            _ => sender.agent_key = "other-agent".into(),
+            _ => sender.subject_id = "subject-b".into(),
         };
         assert!(apply(&conn, &input, &sender).is_err());
     }
     let mut other = source();
     other.worker_id = "replacement".into();
-    // Backend authenticated the replacement as the current stable agent. A new
+    // Backend authenticated the replacement as the current Subject body. A new
     // operation can proceed on the same assignment without starting a new row.
     let next = update(
         "a",
@@ -323,7 +332,7 @@ fn legacy(original: &str, fenced: bool, later_case: bool, reassigned: bool) -> C
     conn.execute_batch("CREATE TABLE ticket_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL); INSERT INTO ticket_schema_migrations VALUES(10,'ticket schema baseline','original-time');").unwrap();
     install_assignment_core(&conn).unwrap();
     seed_ticket(&conn, "w", "t");
-    assign(&conn, "a", "assigned-a");
+    assign_legacy(&conn, "a", "assigned-a");
     let snapshot = TicketLegacyWorkflowSnapshot {
         workflow_state: original.into(),
         workflow_state_explicit: true,
@@ -347,7 +356,7 @@ fn legacy(original: &str, fenced: bool, later_case: bool, reassigned: bool) -> C
         conn.execute("INSERT INTO typed_ticket_case_history VALUES('w','t',0,'closed-later','case-fingerprint','{}')",[]).unwrap();
     }
     if reassigned {
-        assign(&conn, "b", "assigned-b");
+        assign_legacy(&conn, "b", "assigned-b");
     }
     conn
 }
@@ -408,7 +417,7 @@ fn migration_preserves_original_records_and_only_reconciles_current_fenced_deleg
                 |r| r.get::<_, i64>(0)
             )
             .unwrap(),
-            13
+            14
         );
         let value = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
             .unwrap()
@@ -459,8 +468,8 @@ fn migration_preserves_original_records_and_only_reconciles_current_fenced_deleg
 fn tracking_schema_rejects_drift_and_duplicate_current_singleton_roles() {
     let conn = db();
     assign(&conn, "a", "assigned-a");
-    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','author','now')",[]).unwrap();
-    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','now')",[]).is_err());
+    conn.execute("INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','b','orchestrator','subject','subject-a','author','now')",[]).unwrap();
+    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','subject','subject-a','now')",[]).is_err());
     conn.execute_batch("DROP TRIGGER typed_ticket_assignment_tracking_history_no_update; CREATE TRIGGER typed_ticket_assignment_tracking_history_no_update BEFORE UPDATE ON typed_ticket_assignment_tracking_history BEGIN SELECT 1; END;").unwrap();
     assert!(
         verify_sqlite_ticket_schema(&conn)
@@ -493,7 +502,7 @@ fn standalone_assignment_identity_audit_and_terminal_receipts_are_immutable_but_
  {
     let conn = db();
     assign(&conn, "a", "assigned-a");
-    conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,principal_id,assignment_id,created_at,request_fingerprint) VALUES('w','assigned-op','assign','t','orchestrator','workspace_agent','workspace-orchestrator','a','original-time','original-fingerprint')",[]).unwrap();
+    conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,principal_id,assignment_id,created_at,request_fingerprint) VALUES('w','assigned-op','assign','t','orchestrator','subject','subject-a','a','original-time','original-fingerprint')",[]).unwrap();
     for sql in [
         "UPDATE ticket_worker_assignments SET assignment_id='changed' WHERE assignment_id='a'",
         "UPDATE ticket_worker_assignments SET assigned_by='forged' WHERE assignment_id='a'",
@@ -518,7 +527,7 @@ fn standalone_assignment_identity_audit_and_terminal_receipts_are_immutable_but_
     // current row, even in a standalone Ticket store without host constraints.
     conn.execute("DELETE FROM ticket_current_worker_assignments", [])
         .unwrap();
-    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','a','orchestrator','workspace_agent','different-agent','now')",[]).is_err());
+    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','a','orchestrator','subject','subject-b','now')",[]).is_err());
     conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,runtime_id,worker_id,created_at,request_fingerprint,claim_state) VALUES('w','pending','assign','t','worker','worker','runtime','pending-worker','claim-time','claim-fingerprint','pending')",[]).unwrap();
     conn.execute("UPDATE ticket_assignment_operations SET worker_id='resolved-worker',binding_recovery_json='{}' WHERE operation_id='pending'",[]).unwrap();
     conn.execute("UPDATE ticket_assignment_operations SET claim_state='failed',failure_reason='Admission failed' WHERE operation_id='pending'",[]).unwrap();
@@ -573,7 +582,7 @@ fn release_projection_preserves_actor_time_reason_and_optional_legacy_tracking()
         TicketAssignmentTrackingStatus::Waiting
     );
     assert!(conn.execute("UPDATE ticket_worker_assignments SET tracking_json='{\"status\":\"done\"}' WHERE assignment_id='b'",[]).is_err());
-    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','workspace_agent','workspace-orchestrator','now')",[]).is_err());
+    assert!(conn.execute("INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('w','t','b','orchestrator','subject','subject-a','now')",[]).is_err());
     let legacy = legacy("done", true, false, false);
     migrate_sqlite_ticket_schema(&legacy).unwrap();
     legacy
@@ -694,15 +703,15 @@ fn standalone_assignment_constraints_reject_cross_ticket_principals_or_orphan_au
     let conn = db();
     assign(&conn, "a", "assigned-a");
     for sql in [
-        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('other','t','cross-workspace','orchestrator','workspace_agent','workspace-orchestrator','author','now')",
+        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('other','t','cross-workspace','orchestrator','subject','subject-a','author','now')",
         "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','bad-orchestrator','orchestrator','user','user','author','now')",
         "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,runtime_id,worker_id,assigned_by,assigned_at) VALUES('w','t','bad-owner','owner','worker','r','worker','author','now')",
-        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','bad-worker','worker','workspace_agent','workspace-orchestrator','author','now')",
+        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','bad-worker','worker','subject','subject-a','author','now')",
         "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,assigned_by,assigned_at) VALUES('w','t','missing-worker-identity','worker','worker','author','now')",
         "INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at) VALUES('other','t','orchestrator','cross-audit','assigned','a','author','now')",
         "INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at) VALUES('w','missing','orchestrator','cross-ticket-audit','assigned','a','author','now')",
         "INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,created_at) VALUES('other','orphan-operation','assign','t','now')",
-        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at,tracking_json) VALUES('w','t','bad-status','orchestrator','workspace_agent','workspace-orchestrator','author','now','{\"status\":\"running\"}')",
+        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at,tracking_json) VALUES('w','t','bad-status','orchestrator','subject','subject-a','author','now','{\"status\":\"running\"}')",
     ] {
         assert!(conn.execute(sql, []).is_err(), "reject {sql}");
     }
@@ -850,3 +859,269 @@ fn assignment_answer_rejects_unbounded_body_and_malformed_identities_without_app
         0
     );
 }
+
+// Frozen component13 composition, not a latest-schema fixture relabelled as old.
+fn legacy_v13() -> Connection {
+    let conn = legacy("no-snapshot", false, false, false);
+    migrate_assignment_tracking(&conn).unwrap();
+    conn.execute_batch(include_str!("post_migration.sql"))
+        .unwrap();
+    conn.execute_batch(include_str!("content_changes.sql"))
+        .unwrap();
+    conn.execute_batch(
+        "INSERT INTO ticket_schema_migrations VALUES
+        (11,'Ticket-owned assignment tracking','v11-time'),
+        (12,'Canonical Ticket posts','v12-time'),
+        (13,'Ticket content change facts','v13-time');",
+    )
+    .unwrap();
+    conn
+}
+
+#[test]
+fn subject_migration_preserves_legacy_audit_and_receipts_without_inventing_subject_authority() {
+    let conn = legacy_v13();
+    let tracking = r#"{"status":"done","waiting_reason":null,"progress":null,"next_action":null,"result":"Historical result"}"#;
+    let old_source = r#"{"actor":"old-actor","runtime_id":"old-runtime","worker_id":"old-worker","agent_key":"workspace-orchestrator","authentication_binding_id":"old-binding"}"#;
+    conn.execute("UPDATE ticket_worker_assignments SET tracking_json=?1,last_tracking_event_id='old-event',tracking_updated_at='old-time' WHERE assignment_id='a'",[tracking]).unwrap();
+    conn.execute("INSERT INTO typed_ticket_assignment_tracking_history VALUES('w','t','a','old-event','old-op',?1,?2,'old-time')",params![tracking,old_source]).unwrap();
+    let before = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
+        .unwrap()
+        .unwrap();
+    let history = sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a").unwrap();
+    // Pre-Subject saved results remain byte-identical and are not new requests.
+    let mut old_input = update(
+        "a",
+        "assigned-a",
+        "old-op",
+        TicketAssignmentTrackingStatus::Done,
+    );
+    old_input.tracking = before.tracking.clone().unwrap();
+    conn.execute(
+        "INSERT INTO typed_ticket_assignment_tracking_operations VALUES('w','old-op','t',?1,?2,?3)",
+        params![
+            serde_json::to_string(&("t", &old_input)).unwrap(),
+            old_source,
+            serde_json::to_string(&before).unwrap()
+        ],
+    )
+    .unwrap();
+    conn.execute("INSERT INTO ticket_assignment_operations(workspace_id,operation_id,action,ticket_id,role,principal_kind,principal_id,assignment_id,created_at,request_fingerprint) VALUES('w','legacy-assign','assign','t','orchestrator','workspace_agent','workspace-orchestrator','a','old-time','old-fingerprint')",[]).unwrap();
+    let receipts = |db: &Connection| -> (String, String) {
+        db.query_row("SELECT source_context_json,result_json FROM typed_ticket_assignment_tracking_operations WHERE operation_key='old-op'",[],|r|Ok((r.get(0)?,r.get(1)?))).unwrap()
+    };
+    let saved_receipt = receipts(&conn);
+    migrate_sqlite_ticket_schema(&conn).unwrap();
+    let after = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        after.principal,
+        TicketAssignmentPrincipalSnapshot::WorkspaceAgent {
+            agent_key: "workspace-orchestrator".into()
+        }
+    );
+    assert_eq!(after.tracking, before.tracking);
+    assert_eq!(after.last_event_id, before.last_event_id);
+    assert_eq!(after.updated_at, before.updated_at);
+    assert!(after.current);
+    assert!(
+        after
+            .migration_diagnostic
+            .as_ref()
+            .unwrap()
+            .contains("nonoperational")
+    );
+    assert_eq!(
+        sqlite_ticket_assignment_tracking_history(&conn, "w", "t", "a").unwrap(),
+        history
+    );
+    assert_eq!(
+        history.last().unwrap().source_context["agent_key"],
+        "workspace-orchestrator"
+    );
+    assert_eq!(receipts(&conn), saved_receipt);
+    assert_eq!(conn.query_row("SELECT principal_kind FROM ticket_assignment_operations WHERE operation_id='legacy-assign'",[],|r|r.get::<_,String>(0)).unwrap(),"workspace_agent");
+    // Neither a random Subject nor spelling the old singleton as a Subject owns it.
+    for subject in ["subject-a", "workspace-orchestrator"] {
+        let mut sender = source();
+        sender.actor = "old-actor".into();
+        sender.runtime_id = "old-runtime".into();
+        sender.worker_id = "old-worker".into();
+        sender.authentication_binding_id = "old-binding".into();
+        sender.subject_id = subject.into();
+        assert!(
+            apply(
+                &conn,
+                &update(
+                    "a",
+                    "old-event",
+                    "new-op",
+                    TicketAssignmentTrackingStatus::Inprogress
+                ),
+                &sender
+            )
+            .is_err()
+        );
+        // Exact historical input and concrete sender still cannot reinterpret
+        // old agent_key JSON as a new Subject-authenticated receipt.
+        assert!(apply(&conn, &old_input, &sender).is_err());
+    }
+    for sql in [
+        "UPDATE ticket_worker_assignments SET tracking_updated_at='new-time' WHERE assignment_id='a'",
+        "INSERT INTO typed_ticket_assignment_tracking_history VALUES('w','t','a','new-event','new-op','{\"status\":\"done\"}','{}','now')",
+        "INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('w','t','legacy-new','orchestrator','workspace_agent','workspace-orchestrator','actor','now')",
+    ] {
+        assert!(conn.execute(sql, []).is_err(), "reject {sql}");
+    }
+    migrate_sqlite_ticket_schema(&conn).unwrap();
+    assert_eq!(
+        sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
+            .unwrap()
+            .unwrap(),
+        after
+    );
+    // Explicit reissue is a new Subject identity, not a conversion of old audit.
+    assign(&conn, "subject-reissue", "subject-reissue-event");
+    apply(
+        &conn,
+        &update(
+            "subject-reissue",
+            "subject-reissue-event",
+            "subject-update",
+            TicketAssignmentTrackingStatus::Done,
+        ),
+        &source(),
+    )
+    .unwrap();
+    assert_eq!(receipts(&conn), saved_receipt);
+    assert!(
+        !sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
+            .unwrap()
+            .unwrap()
+            .current
+    );
+    let wire = serde_json::to_value(source()).unwrap();
+    assert_eq!(wire["subject_id"], "subject-a");
+    assert!(wire.get("agent_key").is_none());
+    // Old serialized principal snapshots and audit JSON are still readable as old types.
+    let old: TicketAssignmentTrackingProjection = serde_json::from_str(&saved_receipt.1).unwrap();
+    assert_eq!(old.principal, before.principal);
+}
+
+#[test]
+fn subject_owner_is_required_for_new_operations_and_worker_replacement_is_not_reassignment() {
+    let conn = db();
+    assign(&conn, "a", "assigned-a");
+    let input = update(
+        "a",
+        "assigned-a",
+        "fresh",
+        TicketAssignmentTrackingStatus::Inprogress,
+    );
+    let mut wrong_subject = source();
+    wrong_subject.subject_id = "subject-b".into();
+    assert!(apply(&conn, &input, &wrong_subject).is_err());
+    let projection = sqlite_ticket_assignment_tracking(&conn, "w", "t", "a")
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        projection.principal,
+        TicketAssignmentPrincipalSnapshot::Subject {
+            subject_id: "subject-a".into()
+        }
+    );
+    let mut replacement = source();
+    replacement.worker_id = "new-body".into();
+    replacement.authentication_binding_id = "new-binding".into();
+    let result = apply(&conn, &input, &replacement).unwrap();
+    assert_eq!(result.assignment_id, "a");
+    assert_eq!(
+        conn.query_row(
+            "SELECT COUNT(*) FROM typed_ticket_assignment_tracking_operations",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        1
+    );
+}
+
+#[test]
+fn failed_subject_migration_rolls_back_schema_and_receipt_but_preserves_host_transaction() {
+    let conn = legacy_v13();
+    let original: String = conn
+        .query_row(
+            "SELECT sql FROM sqlite_schema WHERE name='ticket_worker_assignments'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap();
+    conn.execute_batch("CREATE TRIGGER host_reject_subject_migration BEFORE UPDATE OF tracking_migration_diagnostic ON ticket_worker_assignments BEGIN SELECT RAISE(ABORT,'simulated failure after CHECK widening'); END;").unwrap();
+    conn.execute_batch("BEGIN IMMEDIATE; CREATE TABLE host_pending(value TEXT); INSERT INTO host_pending VALUES('keep');").unwrap();
+    assert!(migrate_sqlite_ticket_schema_in_transaction(&conn).is_err());
+    assert!(!conn.is_autocommit());
+    assert_eq!(
+        conn.query_row("SELECT value FROM host_pending", [], |r| r
+            .get::<_, String>(0))
+            .unwrap(),
+        "keep"
+    );
+    assert_eq!(
+        conn.query_row(
+            "SELECT MAX(version) FROM ticket_schema_migrations",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        13
+    );
+    assert_eq!(
+        conn.query_row(
+            "SELECT sql FROM sqlite_schema WHERE name='ticket_worker_assignments'",
+            [],
+            |r| r.get::<_, String>(0)
+        )
+        .unwrap(),
+        original
+    );
+    assert!(
+        !conn
+            .pragma_query_value(None, "writable_schema", |r| r.get::<_, bool>(0))
+            .unwrap()
+    );
+    assert_eq!(
+        conn.query_row(
+            "SELECT COUNT(*) FROM sqlite_schema WHERE name='ticket_subject_current_insert'",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        0
+    );
+    conn.execute_batch("DROP TRIGGER host_reject_subject_migration;")
+        .unwrap();
+    migrate_sqlite_ticket_schema_in_transaction(&conn).unwrap();
+    conn.execute_batch("COMMIT").unwrap();
+    verify_sqlite_ticket_schema(&conn).unwrap();
+}
+
+#[test]
+fn subject_generation_is_ledger_selected_and_never_repaired_by_a_marker_trigger() {
+    let conn = legacy_v13();
+    conn.execute_batch("CREATE TRIGGER ticket_subject_current_insert BEFORE INSERT ON ticket_current_worker_assignments BEGIN SELECT 1; END;").unwrap();
+    assert!(migrate_sqlite_ticket_schema(&conn).is_err());
+    assert_eq!(
+        conn.query_row(
+            "SELECT MAX(version) FROM ticket_schema_migrations",
+            [],
+            |r| r.get::<_, i64>(0)
+        )
+        .unwrap(),
+        13
+    );
+    let conn = db();
+    conn.execute_batch("DROP TRIGGER ticket_subject_current_insert; CREATE TRIGGER ticket_subject_current_insert BEFORE INSERT ON ticket_current_worker_assignments BEGIN SELECT 1; END;").unwrap();
+    assert!(verify_sqlite_ticket_schema(&conn).is_err());
+    assert!(migrate_sqlite_ticket_schema(&conn).is_err());
+}

crates/ticket/src/lib.rs

@@ -31,6 +31,7 @@ mod assignment_tracking;
 pub mod config;
 mod content_changes;
 mod sqlite_schema;
+mod subject_assignment;
 mod tags;
 pub use assignment_tracking::*;
 pub use tags::{
@@ -39,6 +40,7 @@ pub use tags::{
 pub mod tool;
 
 pub use sqlite_schema::{
+    HistoricalTicketSchemaTarget, migrate_historical_sqlite_ticket_schema_in_transaction,
     migrate_sqlite_ticket_schema, migrate_sqlite_ticket_schema_in_transaction,
     verify_legacy_sqlite_ticket_schema, verify_pre_classification_ticket_schema,
     verify_sqlite_ticket_schema,

crates/ticket/src/sqlite_schema.rs

@@ -20,7 +20,7 @@ use crate::{Result, TicketError, sqlite_err};
 
 const MIGRATION_TABLE: &str = "ticket_schema_migrations";
 const MAX_SCHEMA_DIAGNOSTICS: usize = 32;
-const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 13;
+const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 14;
 const BASELINE_MIGRATION_NAME: &str = "ticket schema baseline";
 
 #[derive(Clone, Copy)]
@@ -320,11 +320,60 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
     }
 }
 
-/// Appends assignment, post, and content-change fact migrations without replacing applied markers.
-/// The host owns the surrounding transaction.
-/// A savepoint rolls back this component on failure, without committing host work.
-/// Current component 13 is verified only; historical edits are never backfilled.
+/// Startup-only historical composition targets retained by Workspace migrations.
+/// These are frozen component boundaries, not selectable request schemas or a
+/// way to restore legacy execution authority. Current stores must use the latest
+/// composer and request paths must use read-only verification instead.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum HistoricalTicketSchemaTarget {
+    /// Two-state cases plus tags; no owning assignment tracking yet.
+    Cases10,
+    /// Owning assignment tracking, before canonical posts.
+    Tracking11,
+    /// Canonical posts, before content-change facts.
+    Posts12,
+    /// Content-change facts, before Subject responsibility constraints.
+    Content13,
+}
+impl HistoricalTicketSchemaTarget {
+    fn shape(self) -> TicketSchemaShape {
+        match self {
+            Self::Cases10 => TicketSchemaShape::Composed10,
+            Self::Tracking11 => TicketSchemaShape::Tracking11,
+            Self::Posts12 => TicketSchemaShape::Posts12,
+            Self::Content13 => TicketSchemaShape::ContentChanges13,
+        }
+    }
+}
+
+/// Compose one explicit retained host-migration boundary in the host's startup
+/// transaction. This is NOT a request-path API or a downgrade/repair operation.
+///
+/// A recognized older preimage is verified before composition. Applied receipts
+/// are retained; only missing migration receipts are appended. If the stored
+/// version is already at or above the target, its ACTUAL version and complete
+/// shape are verified without downgrading, relabelling, or replaying old steps.
+/// Unknown history and weakened constraints fail closed. The component savepoint
+/// rolls back failure without committing or discarding the caller's other work.
+pub fn migrate_historical_sqlite_ticket_schema_in_transaction(
+    connection: &Connection,
+    target: HistoricalTicketSchemaTarget,
+) -> Result<()> {
+    migrate_ticket_schema_to_in_transaction(connection, target.shape())
+}
+
+/// Appends owning component migrations through the latest contract without
+/// replacing applied receipts. Startup only; the host owns the transaction.
+/// A savepoint rolls back this component on failure without committing host work.
+/// Already-current schemas are verified, never repaired or backfilled.
 pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> Result<()> {
+    migrate_ticket_schema_to_in_transaction(connection, TicketSchemaShape::Subjects14)
+}
+
+fn migrate_ticket_schema_to_in_transaction(
+    connection: &Connection,
+    target: TicketSchemaShape,
+) -> Result<()> {
     if connection.is_autocommit() {
         return Err(TicketError::Sqlite(
             "Ticket component migration requires a host transaction".into(),
@@ -344,26 +393,36 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
         verify_table(connection, MIGRATION_TABLE, MIGRATION_COLUMNS, &[], false)?;
         let applied = load_applied_migrations(connection)?;
         if applied.is_empty() {
-            create_latest_ticket_schema(connection)?;
+            connection
+                .execute_batch(include_str!("latest_schema.sql"))
+                .map_err(sqlite_err)?;
+            for version in 11..=target.version() {
+                apply_component_migration(connection, version)?;
+            }
             connection
                 .execute(
-                    "INSERT INTO ticket_schema_migrations VALUES (13, ?1, ?2)",
-                    params![BASELINE_MIGRATION_NAME, chrono::Utc::now().to_rfc3339()],
+                    "INSERT INTO ticket_schema_migrations VALUES (?1, ?2, ?3)",
+                    params![
+                        target.version(),
+                        BASELINE_MIGRATION_NAME,
+                        chrono::Utc::now().to_rfc3339()
+                    ],
                 )
                 .map_err(sqlite_err)?;
         } else {
-            let version = recognized_schema_version(&applied)?;
+            let mut version = recognized_schema_version(&applied)?;
             if version == 6 {
+                // Retained v6 conversion establishes the verified v7 preimage.
+                // Preserve its original baseline receipt instead of renumbering it.
                 migrate_v6_to_v7(connection)?;
-                connection
-                    .execute(
-                        "UPDATE ticket_schema_migrations SET version = 7 WHERE version = 6",
-                        [],
-                    )
-                    .map_err(sqlite_err)?;
+                append_component_receipt(connection, 7)?;
+                version = 7;
             }
-            let shape = TicketSchemaShape::from_version(if version == 6 { 7 } else { version });
+            let shape = TicketSchemaShape::from_version(version);
             verify_ticket_schema_shape(connection, shape)?;
+            if version >= target.version() {
+                return Ok(());
+            }
             match shape {
                 TicketSchemaShape::Legacy7 => {
                     add_tag_schema(connection)?;
@@ -374,30 +433,19 @@ pub fn migrate_sqlite_ticket_schema_in_transaction(connection: &Connection) -> R
                 TicketSchemaShape::Composed10
                 | TicketSchemaShape::Tracking11
                 | TicketSchemaShape::Posts12
-                | TicketSchemaShape::ContentChanges13 => {}
+                | TicketSchemaShape::ContentChanges13
+                | TicketSchemaShape::Subjects14 => {}
             }
             if version < 10 {
-                connection
-                    .execute(
-                        "UPDATE ticket_schema_migrations SET version = 10, applied_at = ?1",
-                        params![chrono::Utc::now().to_rfc3339()],
-                    )
-                    .map_err(sqlite_err)?;
+                append_component_receipt(connection, 10)?;
+                version = 10;
             }
-            if version < 11 {
-                crate::assignment_tracking::migrate_assignment_tracking(connection)?;
-                connection.execute("INSERT INTO ticket_schema_migrations VALUES (11, 'Ticket-owned assignment tracking', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
-            }
-            if version < 12 {
-                migrate_posts(connection)?;
-                connection.execute("INSERT INTO ticket_schema_migrations VALUES (12, 'Canonical Ticket posts', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
-            }
-            if version < 13 {
-                migrate_content_changes(connection)?;
-                connection.execute("INSERT INTO ticket_schema_migrations VALUES (13, 'Ticket content change facts', ?1)", params![chrono::Utc::now().to_rfc3339()]).map_err(sqlite_err)?;
+            for next in version + 1..=target.version() {
+                apply_component_migration(connection, next)?;
+                append_component_receipt(connection, next)?;
             }
         }
-        verify_sqlite_ticket_schema(connection)
+        verify_ticket_schema_shape(connection, target)
     })();
     match result {
         Ok(()) => connection
@@ -421,6 +469,7 @@ enum TicketSchemaShape {
     Tracking11,
     Posts12,
     ContentChanges13,
+    Subjects14,
 }
 impl TicketSchemaShape {
     fn from_version(version: i64) -> Self {
@@ -432,6 +481,7 @@ impl TicketSchemaShape {
             11 => Self::Tracking11,
             12 => Self::Posts12,
             13 => Self::ContentChanges13,
+            14 => Self::Subjects14,
             _ => unreachable!(),
         }
     }
@@ -444,6 +494,7 @@ impl TicketSchemaShape {
             Self::Tracking11 => 11,
             Self::Posts12 => 12,
             Self::ContentChanges13 => 13,
+            Self::Subjects14 => 14,
         }
     }
     fn tags(self) -> bool {
@@ -454,6 +505,7 @@ impl TicketSchemaShape {
                 | Self::Tracking11
                 | Self::Posts12
                 | Self::ContentChanges13
+                | Self::Subjects14
         )
     }
     fn cases(self) -> bool {
@@ -464,13 +516,14 @@ impl TicketSchemaShape {
                 | Self::Tracking11
                 | Self::Posts12
                 | Self::ContentChanges13
+                | Self::Subjects14
         )
     }
 }
 
 /// Verifies the current Ticket-owned SQLite schema without executing DDL.
 pub fn verify_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
-    verify_ticket_schema_shape(connection, TicketSchemaShape::ContentChanges13)
+    verify_ticket_schema_shape(connection, TicketSchemaShape::Subjects14)
 }
 
 /// Retained Workspace schema-66 migration produces the pre-classification v7
@@ -484,7 +537,7 @@ pub fn verify_pre_classification_ticket_schema(connection: &Connection) -> Resul
 /// treating historical workflow values as the current two-state case authority.
 pub fn verify_legacy_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
     let applied = load_applied_migrations(connection)?;
-    let version = applied.keys().next().copied().unwrap_or_default();
+    let version = recognized_schema_version(&applied)?;
     if !matches!(version, 7 | 8) {
         return Err(TicketError::Sqlite(
             "legacy Ticket schema requires component 7 or 8".into(),
@@ -586,7 +639,9 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
         let mut columns = columns.to_vec();
         if matches!(
             shape,
-            TicketSchemaShape::Posts12 | TicketSchemaShape::ContentChanges13
+            TicketSchemaShape::Posts12
+                | TicketSchemaShape::ContentChanges13
+                | TicketSchemaShape::Subjects14
         ) {
             if table == "typed_tickets" {
                 columns.retain(|c| c.name != "body");
@@ -831,16 +886,20 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
         TicketSchemaShape::Tracking11
             | TicketSchemaShape::Posts12
             | TicketSchemaShape::ContentChanges13
+            | TicketSchemaShape::Subjects14
     ) {
-        if let Err(error) =
-            crate::assignment_tracking::verify_assignment_tracking_schema(connection)
-        {
+        if let Err(error) = crate::assignment_tracking::verify_assignment_tracking_schema(
+            connection,
+            matches!(shape, TicketSchemaShape::Subjects14),
+        ) {
             push_diagnostic(&mut diagnostics, error.to_string());
         }
     }
     if matches!(
         shape,
-        TicketSchemaShape::Posts12 | TicketSchemaShape::ContentChanges13
+        TicketSchemaShape::Posts12
+            | TicketSchemaShape::ContentChanges13
+            | TicketSchemaShape::Subjects14
     ) {
         collect_index_diagnostics(
             connection,
@@ -851,7 +910,10 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
             &mut diagnostics,
         );
     }
-    if matches!(shape, TicketSchemaShape::ContentChanges13) {
+    if matches!(
+        shape,
+        TicketSchemaShape::ContentChanges13 | TicketSchemaShape::Subjects14
+    ) {
         collect_content_change_diagnostics(connection, &mut diagnostics)?;
     }
     if diagnostics.is_empty() {
@@ -870,13 +932,41 @@ fn verify_ticket_schema_shape(connection: &Connection, shape: TicketSchemaShape)
     }
 }
 
-fn create_latest_ticket_schema(connection: &Connection) -> Result<()> {
+// Both fresh and retained composition use this single ordered step dispatcher.
+fn apply_component_migration(connection: &Connection, version: i64) -> Result<()> {
+    match version {
+        11 => crate::assignment_tracking::migrate_assignment_tracking(connection),
+        12 => migrate_posts(connection),
+        13 => migrate_content_changes(connection),
+        14 => crate::subject_assignment::migrate(connection),
+        _ => unreachable!("bounded component migration"),
+    }
+}
+
+fn component_migration_name(version: i64) -> &'static str {
+    match version {
+        7 => "Ordered Ticket targets",
+        10 => "Ticket case and tag composition",
+        11 => "Ticket-owned assignment tracking",
+        12 => "Canonical Ticket posts",
+        13 => "Ticket content change facts",
+        14 => "Subject Ticket responsibility",
+        _ => unreachable!("bounded component receipt"),
+    }
+}
+
+fn append_component_receipt(connection: &Connection, version: i64) -> Result<()> {
     connection
-        .execute_batch(include_str!("latest_schema.sql"))
+        .execute(
+            "INSERT INTO ticket_schema_migrations VALUES (?1, ?2, ?3)",
+            params![
+                version,
+                component_migration_name(version),
+                chrono::Utc::now().to_rfc3339()
+            ],
+        )
         .map_err(sqlite_err)?;
-    crate::assignment_tracking::migrate_assignment_tracking(connection)?;
-    migrate_posts(connection)?;
-    migrate_content_changes(connection)
+    Ok(())
 }
 
 fn migrate_content_changes(connection: &Connection) -> Result<()> {
@@ -1212,26 +1302,24 @@ fn load_applied_migrations(connection: &Connection) -> Result<BTreeMap<i64, Stri
 }
 
 fn recognized_schema_version(applied: &BTreeMap<i64, String>) -> Result<i64> {
-    if applied.len() == 1 {
-        if let Some((&version, name)) = applied.first_key_value() {
-            if (6..=13).contains(&version) && name == BASELINE_MIGRATION_NAME {
-                return Ok(version);
-            }
-        }
-    }
-    for baseline in [10, 11, 12] {
-        let mut expected = BTreeMap::from([(baseline, BASELINE_MIGRATION_NAME.to_owned())]);
-        for (version, name) in [
-            (11, "Ticket-owned assignment tracking"),
-            (12, "Canonical Ticket posts"),
-            (13, "Ticket content change facts"),
-        ] {
-            if version <= baseline {
-                continue;
-            }
-            expected.insert(version, name.to_owned());
+    if let Some((&baseline, name)) = applied.first_key_value() {
+        if (6..=LATEST_SQLITE_TICKET_SCHEMA_VERSION).contains(&baseline)
+            && name == BASELINE_MIGRATION_NAME
+        {
+            let mut expected = BTreeMap::from([(baseline, name.clone())]);
             if applied == &expected {
-                return Ok(version);
+                return Ok(baseline);
+            }
+            // Versions 8 and 9 were independent baseline generations, not
+            // ordered upgrades. Both compose at10; v6 additionally needs v7.
+            for version in [7, 10, 11, 12, 13, 14] {
+                if version <= baseline || (version == 7 && baseline != 6) {
+                    continue;
+                }
+                expected.insert(version, component_migration_name(version).to_owned());
+                if applied == &expected {
+                    return Ok(version);
+                }
             }
         }
     }
@@ -1733,10 +1821,13 @@ mod tests {
         assert_eq!(
             load_applied_migrations(&connection).unwrap(),
             BTreeMap::from([
-                (10, BASELINE_MIGRATION_NAME.to_owned()),
+                (6, BASELINE_MIGRATION_NAME.to_owned()),
+                (7, "Ordered Ticket targets".to_owned()),
+                (10, "Ticket case and tag composition".to_owned()),
                 (11, "Ticket-owned assignment tracking".to_owned()),
                 (12, "Canonical Ticket posts".to_owned()),
-                (13, "Ticket content change facts".to_owned())
+                (13, "Ticket content change facts".to_owned()),
+                (14, "Subject Ticket responsibility".to_owned())
             ])
         );
     }
@@ -1879,7 +1970,7 @@ mod tests {
 
         let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
         assert!(error.to_string().contains(
-            "migration history must contain a canonical baseline and ordered migrations through 13"
+            "migration history must contain a canonical baseline and ordered migrations through 14"
         ));
         assert_eq!(load_applied_migrations(&connection).unwrap().len(), 2);
     }
@@ -1901,7 +1992,7 @@ mod tests {
 
         let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
         assert!(error.to_string().contains(
-            "migration history must contain a canonical baseline and ordered migrations through 13"
+            "migration history must contain a canonical baseline and ordered migrations through 14"
         ));
         assert!(!table_exists(&connection, "typed_tickets").unwrap());
     }

crates/ticket/src/subject_assignment.rs

@@ -0,0 +1,294 @@
+//! Component 14 owns Subject assignment table constraints and guards. It never
+//! maps a legacy agent to a Subject, starts work, or rewrites historical receipts.
+//! The host separately owns authenticated live Subject bindings, delivery and startup fences.
+use rusqlite::{Connection, OptionalExtension, params};
+
+const PRINCIPAL_TABLES: [&str; 3] = [
+    "ticket_assignment_operations",
+    "ticket_current_worker_assignments",
+    "ticket_worker_assignments",
+];
+const OLD_KINDS: &str = "CHECK(principal_kind IN ('user', 'worker', 'workspace_agent'))";
+const SUBJECT_KINDS: &str =
+    "CHECK(principal_kind IN ('user', 'worker', 'workspace_agent', 'subject'))";
+pub(crate) const SUBJECT_GUARDS: &str = include_str!("subject_assignment_schema.sql");
+const LEGACY_DIAGNOSTIC: &str = "Legacy Orchestrator identity retained without a trustworthy historical Subject mapping; nonoperational, explicit authorized Subject assignment required. Saved history and receipts do not grant execution authority.";
+
+#[derive(Debug, thiserror::Error)]
+enum MigrationError {
+    #[error("{0}")]
+    Sqlite(#[from] rusqlite::Error),
+    #[error("{0}")]
+    Contract(String),
+}
+type Result<T> = std::result::Result<T, MigrationError>;
+
+/// Called only by the owning startup migration, within its savepoint. The caller
+/// rolls back on error. No independent ledger, transaction, or authority fallback.
+pub(crate) fn migrate(conn: &Connection) -> crate::Result<()> {
+    let result = (|| -> Result<()> {
+        if conn.is_autocommit() {
+            return Err(MigrationError::Contract(
+                "Subject assignment migration requires a caller-owned transaction".into(),
+            ));
+        }
+        widen_principal_checks(conn)?;
+        conn.execute_batch(SUBJECT_GUARDS)?;
+        conn.execute(
+            "UPDATE ticket_worker_assignments
+            SET tracking_migration_diagnostic = CASE
+              WHEN tracking_migration_diagnostic IS NULL OR tracking_migration_diagnostic=''
+                THEN ?1 ELSE tracking_migration_diagnostic || char(10) || ?1 END
+            WHERE role='orchestrator' AND principal_kind!='subject'",
+            [LEGACY_DIAGNOSTIC],
+        )?;
+        verify_inner(conn)?;
+        verify_integrity(conn)
+    })();
+    result.map_err(super::sqlite_err)
+}
+
+pub(crate) fn verify(conn: &Connection) -> crate::Result<()> {
+    verify_inner(conn).map_err(super::sqlite_err)
+}
+
+fn verify_inner(conn: &Connection) -> Result<()> {
+    require_schema_writes_closed(conn)?;
+    let canonical = known_table_shape(conn, true)?;
+    canonical.execute_batch(SUBJECT_GUARDS)?;
+    let mut statement =
+        canonical.prepare("SELECT name FROM sqlite_schema WHERE type='trigger' ORDER BY name")?;
+    for name in statement.query_map([], |row| row.get::<_, String>(0))? {
+        require_same_sql(conn, &canonical, "trigger", &name?)?;
+    }
+    Ok(())
+}
+
+fn require_schema_writes_closed(conn: &Connection) -> Result<()> {
+    for setting in ["writable_schema", "ignore_check_constraints"] {
+        if conn.pragma_query_value(None, setting, |row| row.get::<_, bool>(0))? {
+            return Err(MigrationError::Contract(format!(
+                "Subject assignment schema requires {setting} OFF"
+            )));
+        }
+    }
+    Ok(())
+}
+
+fn schema_sql(conn: &Connection, kind: &str, name: &str) -> Result<String> {
+    conn.query_row(
+        "SELECT sql FROM main.sqlite_schema WHERE type=?1 AND name=?2",
+        params![kind, name],
+        |row| row.get(0),
+    )
+    .optional()?
+    .ok_or_else(|| MigrationError::Contract(format!("Missing Subject assignment {kind} {name}")))
+}
+
+fn require_same_sql(
+    conn: &Connection,
+    canonical: &Connection,
+    kind: &str,
+    name: &str,
+) -> Result<()> {
+    if normalize_sql(&schema_sql(conn, kind, name)?)
+        != normalize_sql(&schema_sql(canonical, kind, name)?)
+    {
+        return Err(MigrationError::Contract(format!(
+            "Subject assignment {kind} {name} differs from canonical schema"
+        )));
+    }
+    Ok(())
+}
+
+/// Token boundaries and quoted literal bytes are significant. Only whitespace,
+/// unquoted ASCII case and SQLite's ALTER-generated quoting of these three known
+/// table identifiers are normalized. In particular, do not globally strip quotes,
+/// comments, IF NOT EXISTS, whitespace in literals, or merge adjacent SQL tokens.
+fn normalize_sql(sql: &str) -> Vec<String> {
+    let mut tokens = Vec::new();
+    let mut chars = sql.chars().peekable();
+    while let Some(ch) = chars.next() {
+        if ch.is_whitespace() {
+            continue;
+        }
+        if matches!(ch, '\'' | '"' | '`' | '[') {
+            let end = if ch == '[' { ']' } else { ch };
+            let mut token = String::from(ch);
+            while let Some(next) = chars.next() {
+                token.push(next);
+                if next == end {
+                    if chars.peek() == Some(&end) && end != ']' {
+                        token.push(chars.next().expect("peeked quote"));
+                    } else {
+                        break;
+                    }
+                }
+            }
+            if ch == '"'
+                && PRINCIPAL_TABLES
+                    .iter()
+                    .any(|table| token == format!("\"{table}\""))
+            {
+                token = token[1..token.len() - 1].to_owned();
+            }
+            tokens.push(token);
+        } else if ch.is_alphanumeric() || matches!(ch, '_' | '$') {
+            let mut token = ch.to_string();
+            while chars
+                .peek()
+                .is_some_and(|next| next.is_alphanumeric() || matches!(*next, '_' | '$'))
+            {
+                token.push(chars.next().expect("peeked token"));
+            }
+            tokens.push(token.to_ascii_lowercase());
+        } else {
+            tokens.push(ch.to_string());
+        }
+    }
+    tokens
+}
+
+/// Reconstruct both explicitly supported historical generations from the immutable
+/// Ticket-owned core and tracking DDL. Only Workspace-parent FK presence differs.
+fn canonical_tables(subject: bool, workspace_parent: bool) -> Result<Connection> {
+    let canonical = Connection::open_in_memory()?;
+    let historical = super::TICKET_ASSIGNMENT_SCHEMA_SQL;
+    let mut sql = if workspace_parent {
+        historical.to_owned()
+    } else {
+        historical.replace("FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,", "")
+            .replace(",\n            FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE\n", "\n")
+    };
+    if subject {
+        sql = sql.replace(OLD_KINDS, SUBJECT_KINDS);
+    }
+    canonical.execute_batch(&sql)?;
+    canonical.execute_batch(include_str!("assignment_tracking_schema.sql"))?;
+    Ok(canonical)
+}
+
+/// Match a whole published three-table generation. The operation table selects
+/// the known Workspace-FK variant; the other two must match that SAME variant.
+/// Mixed principal generations or one-off dropped constraints are not accepted.
+fn known_table_shape(conn: &Connection, subject: bool) -> Result<Connection> {
+    let operation = normalize_sql(&schema_sql(conn, "table", PRINCIPAL_TABLES[0])?);
+    for workspace_parent in [true, false] {
+        let canonical = canonical_tables(subject, workspace_parent)?;
+        if operation != normalize_sql(&schema_sql(&canonical, "table", PRINCIPAL_TABLES[0])?) {
+            continue;
+        }
+        for table in PRINCIPAL_TABLES {
+            if normalize_sql(&schema_sql(conn, "table", table)?)
+                != normalize_sql(&schema_sql(&canonical, "table", table)?)
+            {
+                return Err(MigrationError::Contract(format!(
+                    "Unsupported complete assignment table SQL in {table}; refusing unknown or mixed schema"
+                )));
+            }
+        }
+        return Ok(canonical);
+    }
+    Err(MigrationError::Contract(format!(
+        "Unsupported complete assignment table SQL in {}; refusing unknown schema",
+        PRINCIPAL_TABLES[0]
+    )))
+}
+
+/// SQLite's documented CHECK-only schema-change procedure avoids rebuilding
+/// immutable assignment parents and their many FK dependents. No columns,
+/// rootpages, rowids, index definitions, or data bytes change. The *entire* known
+/// old/new table SQL is preflighted before schema writes. The UPDATE targets are
+/// exactly the three named main-schema table rows and use exact-SQL CAS.
+///
+/// A normal DROP/rebuild needs foreign_keys=OFF *before* the caller transaction:
+/// defer_foreign_keys does not defer CASCADE actions or immutable child-delete
+/// triggers. Renaming with FK enforcement rewrites inbound references. Rebuilding
+/// the complete FK graph would include immutable tracking, receipts, work releases,
+/// origins, deliveries, startup fences and historical implementation delegations.
+/// That would require a different migration boundary and much broader data churn.
+/// For this CHECK-only change under the agreed caller transaction, the documented
+/// bounded schema edit is safer than an improvised graph rebuild. If table layout
+/// ever changes, this helper must reject it; it is not a general migration engine.
+fn widen_principal_checks(conn: &Connection) -> Result<()> {
+    require_schema_writes_closed(conn)?;
+    let old = known_table_shape(conn, false)?;
+    let mut changes = Vec::new();
+    for table in PRINCIPAL_TABLES {
+        let sql = schema_sql(conn, "table", table)?;
+        if sql.matches(OLD_KINDS).count() != 1 {
+            return Err(MigrationError::Contract(format!(
+                "Unsupported principal CHECK spelling in {table}"
+            )));
+        }
+        changes.push((table, sql.replace(OLD_KINDS, SUBJECT_KINDS), sql));
+    }
+    // Do not silently repair same-named weakened principal triggers on upgrade.
+    for name in [
+        "typed_ticket_assignment_orchestrator_principal_insert",
+        "typed_ticket_assignment_role_principal_insert",
+    ] {
+        require_same_sql(conn, &old, "trigger", name)?;
+    }
+    let cookie: i64 = conn.pragma_query_value(None, "schema_version", |row| row.get(0))?;
+    if !(0..i64::from(i32::MAX)).contains(&cookie) {
+        return Err(MigrationError::Contract(
+            "SQLite schema cookie cannot be advanced".into(),
+        ));
+    }
+    conn.pragma_update(None, "writable_schema", true)?;
+    let change_result = (|| {
+        let enabled: bool = conn.pragma_query_value(None, "writable_schema", |row| row.get(0))?;
+        if !enabled {
+            return Err(MigrationError::Contract(
+                "SQLite refused the CHECK-only Subject schema migration".into(),
+            ));
+        }
+        for (table, new_sql, old_sql) in changes {
+            let changed = conn.execute(
+                "UPDATE main.sqlite_schema SET sql=?2 WHERE type='table' AND name=?1 AND tbl_name=?1 AND sql=?3",
+                params![table, new_sql, old_sql],
+            )?;
+            if changed != 1 {
+                return Err(MigrationError::Contract(format!(
+                    "Principal CHECK changed concurrently in {table}"
+                )));
+            }
+        }
+        // Engine schema cache generation, NOT the application migration ledger.
+        conn.pragma_update(None, "schema_version", cookie + 1)?;
+        Ok(())
+    })();
+    // writable_schema is connection state, not transactional: restore it even
+    // on failure, before returning control to the caller's rollback path.
+    let restore_result = conn.pragma_update(None, "writable_schema", false);
+    restore_result?;
+    require_schema_writes_closed(conn)?;
+    change_result?;
+    conn.flush_prepared_statement_cache();
+    // Force a complete reparse with writable_schema OFF. Malformed schema must
+    // fail here rather than leave an apparently successful migration receipt.
+    for table in PRINCIPAL_TABLES {
+        conn.prepare(&format!("SELECT * FROM {table} LIMIT 0"))?;
+    }
+    Ok(())
+}
+
+fn verify_integrity(conn: &Connection) -> Result<()> {
+    let integrity: String = conn.query_row("PRAGMA integrity_check", [], |row| row.get(0))?;
+    if integrity != "ok" {
+        return Err(MigrationError::Contract(format!(
+            "Subject assignment migration integrity check failed: {integrity}"
+        )));
+    }
+    let violations: i64 =
+        conn.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
+            row.get(0)
+        })?;
+    if violations != 0 {
+        return Err(MigrationError::Contract(
+            "Subject assignment migration foreign-key violation".into(),
+        ));
+    }
+    Ok(())
+}

crates/ticket/src/subject_assignment_schema.sql

@@ -0,0 +1,54 @@
+-- Component 14: additive Subject responsibility guards. Historical identities remain readable.
+
+DROP TRIGGER typed_ticket_assignment_orchestrator_principal_insert;
+CREATE TRIGGER typed_ticket_assignment_orchestrator_principal_insert
+BEFORE INSERT ON ticket_worker_assignments
+WHEN NEW.role='orchestrator' AND NEW.principal_kind!='subject'
+BEGIN SELECT RAISE(ABORT,'New Orchestrator responsibility requires a Subject principal'); END;
+
+DROP TRIGGER typed_ticket_assignment_role_principal_insert;
+CREATE TRIGGER typed_ticket_assignment_role_principal_insert
+BEFORE INSERT ON ticket_worker_assignments
+WHEN (NEW.role='owner' AND NEW.principal_kind!='user')
+ OR (NEW.role='worker' AND NEW.principal_kind IN ('workspace_agent','subject'))
+ OR (NEW.principal_kind='workspace_agent' AND NEW.principal_id!='workspace-orchestrator')
+ OR (NEW.principal_kind='subject' AND NEW.role!='orchestrator')
+BEGIN SELECT RAISE(ABORT,'Assignment role and principal are incompatible'); END;
+
+CREATE TRIGGER ticket_subject_current_insert
+BEFORE INSERT ON ticket_current_worker_assignments
+WHEN (NEW.role='orchestrator' AND NEW.principal_kind!='subject')
+ OR (NEW.principal_kind='subject' AND NEW.role!='orchestrator')
+BEGIN SELECT RAISE(ABORT,'Current Orchestrator responsibility requires a Subject principal'); END;
+CREATE TRIGGER ticket_subject_current_update
+BEFORE UPDATE OF workspace_id,ticket_id,role,assignment_id,principal_kind,principal_id,runtime_id,worker_id
+ON ticket_current_worker_assignments
+WHEN (NEW.role='orchestrator' AND NEW.principal_kind!='subject')
+ OR (NEW.principal_kind='subject' AND NEW.role!='orchestrator')
+BEGIN SELECT RAISE(ABORT,'Current Orchestrator responsibility requires a Subject principal'); END;
+
+CREATE TRIGGER ticket_subject_operation_insert
+BEFORE INSERT ON ticket_assignment_operations
+WHEN (NEW.role='orchestrator' AND NEW.action!='unassign' AND NEW.principal_kind!='subject')
+ OR (NEW.principal_kind='subject' AND (NEW.role!='orchestrator'
+   OR NEW.principal_id IS NULL OR length(trim(NEW.principal_id))=0
+   OR NEW.runtime_id IS NOT NULL OR NEW.worker_id IS NOT NULL))
+BEGIN SELECT RAISE(ABORT,'New Orchestrator assignment operation requires a Subject principal'); END;
+CREATE TRIGGER ticket_subject_operation_update
+BEFORE UPDATE ON ticket_assignment_operations
+WHEN (NEW.role='orchestrator' AND NEW.action!='unassign' AND NEW.principal_kind!='subject')
+ OR (NEW.principal_kind='subject' AND (NEW.role!='orchestrator'
+   OR NEW.principal_id IS NULL OR length(trim(NEW.principal_id))=0
+   OR NEW.runtime_id IS NOT NULL OR NEW.worker_id IS NOT NULL))
+BEGIN SELECT RAISE(ABORT,'New Orchestrator assignment operation requires a Subject principal'); END;
+
+CREATE TRIGGER ticket_legacy_orchestrator_tracking_no_update
+BEFORE UPDATE OF tracking_json,last_tracking_event_id,tracking_updated_at ON ticket_worker_assignments
+WHEN OLD.role='orchestrator' AND OLD.principal_kind!='subject'
+BEGIN SELECT RAISE(ABORT,'Legacy Orchestrator responsibility is nonoperational'); END;
+CREATE TRIGGER ticket_legacy_orchestrator_history_no_insert
+BEFORE INSERT ON typed_ticket_assignment_tracking_history
+WHEN EXISTS(SELECT 1 FROM ticket_worker_assignments a
+ WHERE a.workspace_id=NEW.workspace_id AND a.assignment_id=NEW.assignment_id
+ AND a.role='orchestrator' AND a.principal_kind!='subject')
+BEGIN SELECT RAISE(ABORT,'Legacy Orchestrator responsibility is nonoperational'); END;

crates/ticket/tests/migration.rs

@@ -78,7 +78,17 @@ fn standalone_schema7_migration_preserves_six_state_snapshot_and_evidence_withou
             row.get(0)
         })
         .unwrap();
-    assert_eq!(marker, 10);
+    assert_eq!(marker, 7);
+    assert_eq!(
+        connection
+            .query_row(
+                "SELECT applied_at FROM ticket_schema_migrations WHERE version=7",
+                [],
+                |r| r.get::<_, String>(0)
+            )
+            .unwrap(),
+        "legacy"
+    );
     let after_events: Vec<_> = connection.prepare("SELECT ticket_id, from_state, to_state, reason, author, body FROM typed_ticket_events WHERE event_index >= 0 ORDER BY ticket_id").unwrap().query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, String>(4)?, row.get::<_, String>(5)?))).unwrap().collect::<std::result::Result<_, _>>().unwrap();
     assert_eq!(after_events, historical_events);
     let store = SqliteTicketBackend::open_verified(&path, "workspace").unwrap();
@@ -442,7 +452,7 @@ fn tagged_schema8_composes_cases_preserving_tombstones_membership_context_receip
     verify_sqlite_ticket_schema(&connection).unwrap();
     assert_eq!(
         rows(&connection, "ticket_schema_migrations")[0][0],
-        rusqlite::types::Value::Integer(10)
+        rusqlite::types::Value::Integer(8)
     );
     for (table, expected) in preserved.iter().zip(&before) {
         assert_eq!(&rows(&connection, table), expected, "changed {table}");
@@ -852,3 +862,218 @@ fn retained_workspace_foreign_keys_survive_composition_without_rebuilding_ticket
         .unwrap();
     assert_eq!(failures, 0);
 }
+
+fn compose_historical(connection: &Connection, target: HistoricalTicketSchemaTarget) {
+    connection.execute_batch("BEGIN IMMEDIATE").unwrap();
+    migrate_historical_sqlite_ticket_schema_in_transaction(connection, target).unwrap();
+    assert!(!connection.is_autocommit());
+    connection.execute_batch("COMMIT").unwrap();
+}
+
+fn max_component(connection: &Connection) -> i64 {
+    connection
+        .query_row(
+            "SELECT MAX(version) FROM ticket_schema_migrations",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap()
+}
+
+#[test]
+fn historical_composition_stops_at_each_explicit_boundary_and_preserves_applied_receipts() {
+    use HistoricalTicketSchemaTarget::{Cases10, Content13, Posts12, Tracking11};
+    // Both empty stores and real old schema7 preimages must stop at the chosen
+    // contract, rather than installing the latest Subject schema prematurely.
+    for retained in [false, true] {
+        for (target, version) in [
+            (Cases10, 10),
+            (Tracking11, 11),
+            (Posts12, 12),
+            (Content13, 13),
+        ] {
+            let conn = Connection::open_in_memory().unwrap();
+            conn.execute_batch("PRAGMA foreign_keys=ON").unwrap();
+            if retained {
+                schema7(&conn);
+            }
+            assert!(migrate_historical_sqlite_ticket_schema_in_transaction(&conn, target).is_err());
+            compose_historical(&conn, target);
+            assert_eq!(max_component(&conn), version);
+            let exists = |kind: &str, name: &str| -> bool {
+                conn.query_row(
+                    "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type=?1 AND name=?2)",
+                    params![kind, name],
+                    |r| r.get(0),
+                )
+                .unwrap()
+            };
+            assert_eq!(
+                exists("table", "typed_ticket_assignment_tracking_history"),
+                version >= 11
+            );
+            assert_eq!(exists("index", "typed_ticket_events_ref"), version >= 12);
+            assert_eq!(
+                exists("table", "typed_ticket_content_changes"),
+                version >= 13
+            );
+            assert!(!exists("trigger", "ticket_subject_current_insert"));
+            if retained {
+                assert_eq!(
+                    conn.query_row(
+                        "SELECT name,applied_at FROM ticket_schema_migrations WHERE version=7",
+                        [],
+                        |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
+                    )
+                    .unwrap(),
+                    ("ticket schema baseline".into(), "legacy".into())
+                );
+            }
+            let receipts = rows(&conn, "ticket_schema_migrations");
+            compose_historical(&conn, target);
+            assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+            // Older host hooks verify the actual newer historical generation;
+            // they neither replay older steps nor change its receipts.
+            compose_historical(&conn, Cases10);
+            assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+            assert!(verify_sqlite_ticket_schema(&conn).is_err());
+        }
+    }
+    let conn = Connection::open_in_memory().unwrap();
+    schema7(&conn);
+    let mut prior = rows(&conn, "ticket_schema_migrations");
+    for target in [Cases10, Tracking11, Posts12, Content13] {
+        compose_historical(&conn, target);
+        let after = rows(&conn, "ticket_schema_migrations");
+        assert_eq!(&after[..prior.len()], prior.as_slice());
+        prior = after;
+    }
+}
+
+#[test]
+fn historical_composition_verifies_actual_current_contract_without_downgrade_or_repair() {
+    let conn = Connection::open_in_memory().unwrap();
+    migrate_sqlite_ticket_schema(&conn).unwrap();
+    let receipts = rows(&conn, "ticket_schema_migrations");
+    for target in [
+        HistoricalTicketSchemaTarget::Cases10,
+        HistoricalTicketSchemaTarget::Tracking11,
+        HistoricalTicketSchemaTarget::Posts12,
+        HistoricalTicketSchemaTarget::Content13,
+    ] {
+        compose_historical(&conn, target);
+        assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+        verify_sqlite_ticket_schema(&conn).unwrap();
+    }
+    // A lower requested target cannot hide a weakened actual14 guard.
+    conn.execute_batch("DROP TRIGGER ticket_subject_current_insert; CREATE TRIGGER ticket_subject_current_insert BEFORE INSERT ON ticket_current_worker_assignments BEGIN SELECT 1; END;").unwrap();
+    conn.execute_batch("BEGIN IMMEDIATE").unwrap();
+    assert!(
+        migrate_historical_sqlite_ticket_schema_in_transaction(
+            &conn,
+            HistoricalTicketSchemaTarget::Cases10
+        )
+        .is_err()
+    );
+    assert!(!conn.is_autocommit());
+    assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+    conn.execute_batch("ROLLBACK").unwrap();
+
+    // A claimed13 marker over actual10 must also be rejected, not backfilled.
+    let conn = Connection::open_in_memory().unwrap();
+    compose_historical(&conn, HistoricalTicketSchemaTarget::Cases10);
+    conn.execute("UPDATE ticket_schema_migrations SET version=13", [])
+        .unwrap();
+    let before = rows(&conn, "ticket_schema_migrations");
+    conn.execute_batch("BEGIN IMMEDIATE").unwrap();
+    assert!(
+        migrate_historical_sqlite_ticket_schema_in_transaction(
+            &conn,
+            HistoricalTicketSchemaTarget::Cases10
+        )
+        .is_err()
+    );
+    assert_eq!(rows(&conn, "ticket_schema_migrations"), before);
+    conn.execute_batch("ROLLBACK").unwrap();
+}
+
+#[test]
+fn bounded_content13_reopens_after_failed_subject_upgrade_then_migrates_atomically_to14() {
+    let temp = TempDir::new().unwrap();
+    let path = temp.path().join("retained-content13.db");
+    let (receipts, history, old_table);
+    {
+        let conn = Connection::open(&path).unwrap();
+        conn.execute_batch("PRAGMA foreign_keys=ON").unwrap();
+        compose_historical(&conn, HistoricalTicketSchemaTarget::Content13);
+        // Actual retained13 accepts its historical stable-agent contract. It
+        // has not been created at14 and then merely relabelled as old.
+        conn.execute_batch("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,workflow_state,workflow_state_explicit) VALUES('workspace','t','t','Title','open','task','P2','open',1);
+            INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,event_ref,kind,body) VALUES('workspace','t',-1,'t:-1','comment','Body');
+            INSERT INTO ticket_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,assigned_by,assigned_at) VALUES('workspace','t','old','orchestrator','workspace_agent','workspace-orchestrator','old-author','old-time');
+            INSERT INTO ticket_current_worker_assignments(workspace_id,ticket_id,assignment_id,role,principal_kind,principal_id,updated_at) VALUES('workspace','t','old','orchestrator','workspace_agent','workspace-orchestrator','old-time');
+            INSERT INTO ticket_worker_assignment_events(workspace_id,ticket_id,role,event_id,action,assignment_id,actor,created_at,operation_id) VALUES('workspace','t','orchestrator','old-assigned','assigned','old','old-author','old-time','old-op');
+            CREATE TRIGGER host_reject_subject_migration BEFORE UPDATE OF tracking_migration_diagnostic ON ticket_worker_assignments BEGIN SELECT RAISE(ABORT,'forced post-widen failure'); END;").unwrap();
+        receipts = rows(&conn, "ticket_schema_migrations");
+        history = rows(&conn, "typed_ticket_assignment_tracking_history");
+        old_table = conn
+            .query_row(
+                "SELECT sql FROM sqlite_schema WHERE name='ticket_worker_assignments'",
+                [],
+                |r| r.get::<_, String>(0),
+            )
+            .unwrap();
+        conn.execute_batch("BEGIN IMMEDIATE; CREATE TABLE host_work(value TEXT); INSERT INTO host_work VALUES('keep');").unwrap();
+        assert!(migrate_sqlite_ticket_schema_in_transaction(&conn).is_err());
+        assert!(!conn.is_autocommit());
+        assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+        assert_eq!(
+            rows(&conn, "typed_ticket_assignment_tracking_history"),
+            history
+        );
+        assert_eq!(
+            conn.query_row(
+                "SELECT sql FROM sqlite_schema WHERE name='ticket_worker_assignments'",
+                [],
+                |r| r.get::<_, String>(0)
+            )
+            .unwrap(),
+            old_table
+        );
+        assert!(
+            !conn
+                .pragma_query_value(None, "writable_schema", |r| r.get::<_, bool>(0))
+                .unwrap()
+        );
+        conn.execute_batch("COMMIT").unwrap();
+    }
+    let conn = Connection::open(&path).unwrap();
+    conn.execute_batch("PRAGMA foreign_keys=ON").unwrap();
+    compose_historical(&conn, HistoricalTicketSchemaTarget::Content13);
+    assert_eq!(max_component(&conn), 13);
+    assert_eq!(rows(&conn, "ticket_schema_migrations"), receipts);
+    assert_eq!(
+        conn.query_row("SELECT value FROM host_work", [], |r| r.get::<_, String>(0))
+            .unwrap(),
+        "keep"
+    );
+    conn.execute_batch("DROP TRIGGER host_reject_subject_migration")
+        .unwrap();
+    migrate_sqlite_ticket_schema(&conn).unwrap();
+    verify_sqlite_ticket_schema(&conn).unwrap();
+    assert_eq!(max_component(&conn), 14);
+    assert_eq!(
+        &rows(&conn, "ticket_schema_migrations")[..receipts.len()],
+        receipts.as_slice()
+    );
+    assert_eq!(
+        rows(&conn, "typed_ticket_assignment_tracking_history"),
+        history
+    );
+    drop(conn);
+    let conn = Connection::open(&path).unwrap();
+    let applied14 = rows(&conn, "ticket_schema_migrations");
+    compose_historical(&conn, HistoricalTicketSchemaTarget::Content13);
+    assert_eq!(rows(&conn, "ticket_schema_migrations"), applied14);
+    verify_sqlite_ticket_schema(&conn).unwrap();
+}

crates/ticket/tests/posts.rs

@@ -270,7 +270,7 @@ fn schema12_migrates_body_authors_archived_history_and_stable_references_then_re
             |r| r.get::<_, i64>(0)
         )
         .unwrap(),
-        13
+        14
     );
     backend
         .add_event("known".into(), comment("Next", "author"))

crates/ticket/tests/posts/content_changes.rs

@@ -597,9 +597,9 @@ fn verifier_rejects_weakened_fact_constraints_trigger_and_migration_history() {
         assert!(migrate_sqlite_ticket_schema(&f.conn).is_err(), "{from}");
     }
     for mutation in [
-        "UPDATE ticket_schema_migrations SET name='unrecognized' WHERE version=13",
+        "UPDATE ticket_schema_migrations SET name='unrecognized' WHERE version=14",
         "INSERT INTO ticket_schema_migrations VALUES(12,'Canonical Ticket posts','invalid extra history')",
-        "UPDATE ticket_schema_migrations SET version=14 WHERE version=13",
+        "UPDATE ticket_schema_migrations SET version=15 WHERE version=14",
     ] {
         let f = Fixture::new();
         f.conn.execute_batch(mutation).unwrap();