1use std::io::Write;
9use std::path::{Path, PathBuf};
10
11use lean_ctx_protocol::{AcceptedOutcomeV1, ExecutionReceiptV1, TaskEnvelopeV1};
12use serde::{Deserialize, Serialize};
13use sha2::{Digest, Sha256};
14
15const MAX_ARTIFACT_BYTES: u64 = 16 * 1024 * 1024;
16const RECORD_NAME: &str = "leanctx-v4-task-receipt-migration.json";
17
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19#[serde(deny_unknown_fields)]
20pub struct LegacyArtifactDigestV1 {
21 pub name: String,
22 pub sha256: String,
23 pub bytes: u64,
24}
25
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(deny_unknown_fields)]
28pub struct TaskReceiptMigrationRecordV1 {
29 pub schema_version: u32,
30 pub task_id: String,
31 pub receipt_id: String,
32 pub plan_id: String,
33 pub source_artifacts: Vec<LegacyArtifactDigestV1>,
34 pub task_envelope: String,
35 pub receipt_authority: String,
36 pub canonical_successor: String,
37 pub required_action: String,
38 pub sources_retained_for_rollback: bool,
39}
40
41pub fn rollback_legacy_task_receipt(source_dir: &Path) -> Result<(), String> {
42 reject_symlink(source_dir, "source directory")?;
43 let _lock = lock_task_receipt_migration(source_dir)?;
44 let record_path = source_dir.join(RECORD_NAME);
45 reject_symlink(&record_path, "migration record")?;
46 let bytes =
47 std::fs::read(&record_path).map_err(|error| format!("read migration record: {error}"))?;
48 let record: TaskReceiptMigrationRecordV1 = serde_json::from_slice(&bytes)
49 .map_err(|error| format!("parse migration record: {error}"))?;
50 if record.schema_version != 1
51 || record.task_envelope != "canonical_v1_preserved"
52 || record.receipt_authority != "legacy_non_authoritative"
53 || record.canonical_successor != "ReceiptDocumentV1"
54 || record.required_action != "reexecute_with_canonical_producer"
55 || !record.sources_retained_for_rollback
56 {
57 return Err("migration record is not rollback-compatible".to_string());
58 }
59 let expected = [
60 "baseline.json",
61 "context_plan.json",
62 "execution_receipt.json",
63 "outcome.json",
64 "task_envelope.json",
65 ];
66 if record.source_artifacts.len() != expected.len()
67 || record
68 .source_artifacts
69 .iter()
70 .zip(expected)
71 .any(|(artifact, expected_name)| artifact.name != expected_name)
72 {
73 return Err("migration record source set is incomplete or unordered".to_string());
74 }
75 for artifact in &record.source_artifacts {
76 let source = source_dir.join(&artifact.name);
77 reject_symlink(&source, &artifact.name)?;
78 let source_bytes =
79 std::fs::read(&source).map_err(|error| format!("read {}: {error}", artifact.name))?;
80 let digest = crate::core::agent_identity::hex_encode(&Sha256::digest(&source_bytes));
81 if source_bytes.len() as u64 != artifact.bytes || digest != artifact.sha256 {
82 return Err(format!("{} changed since migration", artifact.name));
83 }
84 }
85 std::fs::remove_file(&record_path)
86 .map_err(|error| format!("remove migration record: {error}"))?;
87 crate::core::atomic_fs::sync_directory(source_dir)
88 .map_err(|error| format!("sync rollback directory: {error}"))
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct TaskReceiptMigrationResult {
93 pub record_path: PathBuf,
94 pub already_migrated: bool,
95 pub record: TaskReceiptMigrationRecordV1,
96}
97
98pub fn migrate_legacy_task_receipt(
99 source_dir: &Path,
100) -> Result<TaskReceiptMigrationResult, String> {
101 let _lock = lock_task_receipt_migration(source_dir)?;
102 let record = assess_legacy_task_receipt(source_dir)?;
103 let record_bytes =
104 serde_json::to_vec_pretty(&record).map_err(|error| format!("encode record: {error}"))?;
105 let record_path = source_dir.join(RECORD_NAME);
106 let already_migrated = if record_path.exists() {
107 reject_symlink(&record_path, "migration record")?;
108 let existing = std::fs::read(&record_path)
109 .map_err(|error| format!("read migration record: {error}"))?;
110 if existing != record_bytes {
111 return Err(format!(
112 "refusing to overwrite non-identical migration record: {}",
113 record_path.display()
114 ));
115 }
116 true
117 } else {
118 create_without_overwrite(&record_path, &record_bytes)?;
119 false
120 };
121
122 Ok(TaskReceiptMigrationResult {
123 record_path,
124 already_migrated,
125 record,
126 })
127}
128
129pub fn assess_legacy_task_receipt(
130 source_dir: &Path,
131) -> Result<TaskReceiptMigrationRecordV1, String> {
132 reject_symlink(source_dir, "source directory")?;
133 if !source_dir.is_dir() {
134 return Err(format!(
135 "task/receipt migration source is not a directory: {}",
136 source_dir.display()
137 ));
138 }
139
140 let task_bytes = read_source(source_dir, "task_envelope.json")?;
141 let receipt_bytes = read_source(source_dir, "execution_receipt.json")?;
142 let task: TaskEnvelopeV1 = serde_json::from_slice(&task_bytes)
143 .map_err(|error| format!("parse task_envelope.json: {error}"))?;
144 task.validate()
145 .map_err(|error| format!("validate task_envelope.json: {error}"))?;
146 let receipt: ExecutionReceiptV1 = serde_json::from_slice(&receipt_bytes)
147 .map_err(|error| format!("parse execution_receipt.json: {error}"))?;
148 receipt
149 .validate()
150 .map_err(|error| format!("validate execution_receipt.json: {error}"))?;
151 if task.task_id != receipt.task_id {
152 return Err("task_id mismatch between task envelope and execution receipt".to_string());
153 }
154
155 let mut source_artifacts = vec![
156 artifact_digest("execution_receipt.json", &receipt_bytes),
157 artifact_digest("task_envelope.json", &task_bytes),
158 ];
159 for name in ["baseline.json", "context_plan.json", "outcome.json"] {
160 let bytes = read_source(source_dir, name)?;
161 validate_source_join(name, &bytes, task.task_id.as_str(), &receipt)?;
162 source_artifacts.push(artifact_digest(name, &bytes));
163 }
164 source_artifacts.sort_by(|left, right| left.name.cmp(&right.name));
165
166 Ok(TaskReceiptMigrationRecordV1 {
167 schema_version: 1,
168 task_id: task.task_id.as_str().to_string(),
169 receipt_id: receipt.receipt_id.as_str().to_string(),
170 plan_id: receipt.plan_id.as_str().to_string(),
171 source_artifacts,
172 task_envelope: "canonical_v1_preserved".to_string(),
173 receipt_authority: "legacy_non_authoritative".to_string(),
174 canonical_successor: "ReceiptDocumentV1".to_string(),
175 required_action: "reexecute_with_canonical_producer".to_string(),
176 sources_retained_for_rollback: true,
177 })
178}
179
180fn read_source(source_dir: &Path, name: &str) -> Result<Vec<u8>, String> {
181 let path = source_dir.join(name);
182 reject_symlink(&path, name)?;
183 let metadata = std::fs::metadata(&path).map_err(|error| format!("stat {name}: {error}"))?;
184 if !metadata.is_file() {
185 return Err(format!("{name} is not a regular file"));
186 }
187 if metadata.len() > MAX_ARTIFACT_BYTES {
188 return Err(format!("{name} exceeds {MAX_ARTIFACT_BYTES} bytes"));
189 }
190 std::fs::read(&path).map_err(|error| format!("read {name}: {error}"))
191}
192
193fn reject_symlink(path: &Path, label: &str) -> Result<(), String> {
194 if path
195 .symlink_metadata()
196 .is_ok_and(|metadata| metadata.file_type().is_symlink())
197 {
198 Err(format!("refusing symlink {label}: {}", path.display()))
199 } else {
200 Ok(())
201 }
202}
203
204fn lock_task_receipt_migration(
205 source_dir: &Path,
206) -> Result<crate::core::migration_lock::MigrationLock, String> {
207 reject_symlink(source_dir, "source directory")?;
208 let lock_path = source_dir.join(".leanctx-v4-task-receipt-migration.lock");
209 reject_symlink(&lock_path, "migration lock")?;
210 crate::core::migration_lock::acquire(&lock_path)
211}
212
213fn artifact_digest(name: &str, bytes: &[u8]) -> LegacyArtifactDigestV1 {
214 LegacyArtifactDigestV1 {
215 name: name.to_string(),
216 sha256: crate::core::agent_identity::hex_encode(&Sha256::digest(bytes)),
217 bytes: bytes.len() as u64,
218 }
219}
220
221fn validate_source_join(
222 name: &str,
223 bytes: &[u8],
224 task_id: &str,
225 receipt: &ExecutionReceiptV1,
226) -> Result<(), String> {
227 if name == "outcome.json" {
228 let outcome: AcceptedOutcomeV1 = serde_json::from_slice(bytes)
229 .map_err(|error| format!("parse outcome.json: {error}"))?;
230 outcome
231 .validate()
232 .map_err(|error| format!("validate outcome.json: {error}"))?;
233 if outcome.task_id.as_str() != task_id {
234 return Err("outcome.json task_id does not match task envelope".to_string());
235 }
236 if receipt.outcome_ref.as_deref() != Some(outcome.outcome_id.as_str()) {
237 return Err("outcome.json outcome_id does not match receipt outcome_ref".to_string());
238 }
239 return Ok(());
240 }
241 let value: serde_json::Value =
242 serde_json::from_slice(bytes).map_err(|error| format!("parse {name}: {error}"))?;
243 let object = value
244 .as_object()
245 .ok_or_else(|| format!("{name} must contain a JSON object"))?;
246 if object
247 .get("schema_version")
248 .and_then(serde_json::Value::as_u64)
249 != Some(1)
250 {
251 return Err(format!("{name} schema_version must be 1"));
252 }
253 if object.get("task_id").and_then(serde_json::Value::as_str) != Some(task_id) {
254 return Err(format!("{name} task_id does not match task envelope"));
255 }
256 Ok(())
257}
258
259fn create_without_overwrite(path: &Path, bytes: &[u8]) -> Result<(), String> {
260 let name = path
261 .file_name()
262 .and_then(|value| value.to_str())
263 .ok_or_else(|| format!("invalid record filename: {}", path.display()))?;
264 let staged = path.with_file_name(format!(".{name}.staged"));
265 reject_symlink(&staged, "staged migration record")?;
266 if staged.exists() {
267 std::fs::remove_file(&staged)
268 .map_err(|error| format!("remove stale staged migration record: {error}"))?;
269 }
270 let mut file = std::fs::OpenOptions::new()
271 .write(true)
272 .create_new(true)
273 .open(&staged)
274 .map_err(|error| format!("create {}: {error}", staged.display()))?;
275 if let Err(error) = file.write_all(bytes).and_then(|()| file.sync_all()) {
276 let _ = std::fs::remove_file(&staged);
277 return Err(format!("write {}: {error}", staged.display()));
278 }
279 std::fs::hard_link(&staged, path)
280 .map_err(|error| format!("publish {} without overwrite: {error}", path.display()))?;
281 std::fs::remove_file(&staged)
282 .map_err(|error| format!("remove staged migration record: {error}"))?;
283 crate::core::atomic_fs::sync_directory(path.parent().unwrap_or(Path::new(".")))
284 .map_err(|error| format!("sync task/receipt migration directory: {error}"))?;
285 Ok(())
286}
287
288#[cfg(test)]
289mod tests {
290 use super::*;
291
292 const TASK: &str = r#"{"schema_version":1,"task_id":"task-1","trace_id":"trace-1","project_id":"project-1","session_id":"session-1","agent_id":"agent-1","complexity":"low","created_at":"2026-08-01T09:00:00Z"}"#;
293 const RECEIPT: &str = r#"{"schema_version":1,"receipt_id":"receipt-1","task_id":"task-1","plan_id":"plan-1","context_balance":{"original_tokens":100,"materialized_tokens":80,"delivered_tokens":60,"provider_billed_tokens":60},"fresh_input_tokens":40,"cached_input_tokens":20,"output_tokens":10,"reasoning_tokens":5,"requested_model":"legacy","selected_model":"legacy","provider":"legacy","model_calls":1,"retries":0,"latency_ms":25,"actual_cost_micros":100,"baseline_cost_micros":150,"avoided_cost_micros":50,"etpao_milli":1000,"outcome_ref":"outcome-1","decision_refs":[],"evidence_refs":[],"signature":"legacy-only"}"#;
294
295 fn fixture() -> tempfile::TempDir {
296 let dir = tempfile::tempdir().unwrap();
297 std::fs::write(dir.path().join("task_envelope.json"), TASK).unwrap();
298 std::fs::write(dir.path().join("execution_receipt.json"), RECEIPT).unwrap();
299 std::fs::write(
300 dir.path().join("baseline.json"),
301 r#"{"schema_version":1,"task_id":"task-1"}"#,
302 )
303 .unwrap();
304 std::fs::write(
305 dir.path().join("context_plan.json"),
306 r#"{"schema_version":1,"task_id":"task-1"}"#,
307 )
308 .unwrap();
309 std::fs::write(
310 dir.path().join("outcome.json"),
311 r#"{"schema_version":1,"outcome_id":"outcome-1","task_id":"task-1","accepted":"accepted","quality_score_milli":900,"signals":{"build":"passed","tests":"passed","lint":"passed","typecheck":"not_run","completion":"passed","pr":"not_run","correction":"not_run","rollback":"not_run","retry":"not_run"},"evidence_refs":[{"schema_version":1,"kind":"QualityMeasurement","uri":"urn:evidence:outcome-1","digest":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","signature_status":"NotSigned"}],"observed_at":"2026-08-01T09:01:00Z"}"#,
312 )
313 .unwrap();
314 dir
315 }
316
317 #[test]
318 fn migration_preserves_sources_and_never_promotes_authority() {
319 let dir = fixture();
320 let first = migrate_legacy_task_receipt(dir.path()).unwrap();
321 assert!(!first.already_migrated);
322 assert_eq!(first.record.task_envelope, "canonical_v1_preserved");
323 assert_eq!(first.record.receipt_authority, "legacy_non_authoritative");
324 assert_eq!(
325 first.record.required_action,
326 "reexecute_with_canonical_producer"
327 );
328 assert_eq!(
329 std::fs::read_to_string(dir.path().join("task_envelope.json")).unwrap(),
330 TASK
331 );
332 let second = migrate_legacy_task_receipt(dir.path()).unwrap();
333 assert!(second.already_migrated);
334 assert_eq!(first.record, second.record);
335 }
336
337 #[test]
338 fn rollback_removes_only_record_after_source_verification() {
339 let dir = fixture();
340 let result = migrate_legacy_task_receipt(dir.path()).unwrap();
341 rollback_legacy_task_receipt(dir.path()).unwrap();
342 assert!(!result.record_path.exists());
343 assert_eq!(
344 std::fs::read_to_string(dir.path().join("task_envelope.json")).unwrap(),
345 TASK
346 );
347 }
348
349 #[test]
350 fn rollback_refuses_changed_source_without_mutation() {
351 let dir = fixture();
352 let result = migrate_legacy_task_receipt(dir.path()).unwrap();
353 std::fs::write(dir.path().join("baseline.json"), b"changed").unwrap();
354 assert!(rollback_legacy_task_receipt(dir.path()).is_err());
355 assert!(result.record_path.exists());
356 }
357
358 #[test]
359 fn migration_recovers_stale_partial_staging_record() {
360 let dir = fixture();
361 let staged = dir.path().join(format!(".{RECORD_NAME}.staged"));
362 std::fs::write(&staged, b"partial").unwrap();
363
364 let migrated = migrate_legacy_task_receipt(dir.path()).unwrap();
365
366 assert!(migrated.record_path.exists());
367 assert!(!staged.exists());
368 }
369
370 #[test]
371 fn all_representative_v3_task_spines_migrate_with_source_digests() {
372 let root =
373 Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/v3/task-spine-v1/tasks");
374 let mut sources: Vec<_> = std::fs::read_dir(root)
375 .unwrap()
376 .map(|entry| entry.unwrap().path())
377 .filter(|path| path.is_dir())
378 .collect();
379 sources.sort();
380 assert_eq!(sources.len(), 10);
381
382 for source in sources {
383 let dir = tempfile::tempdir().unwrap();
384 for name in [
385 "task_envelope.json",
386 "baseline.json",
387 "context_plan.json",
388 "execution_receipt.json",
389 "outcome.json",
390 ] {
391 std::fs::copy(source.join(name), dir.path().join(name)).unwrap();
392 }
393 let result = migrate_legacy_task_receipt(dir.path()).unwrap();
394 assert_eq!(result.record.source_artifacts.len(), 5);
395 assert!(result.record.source_artifacts.iter().all(|artifact| {
396 artifact.sha256.len() == 64
397 && artifact
398 .sha256
399 .chars()
400 .all(|character| character.is_ascii_hexdigit())
401 }));
402 }
403 }
404
405 #[test]
406 fn mismatched_task_or_outcome_fails_without_record() {
407 let dir = fixture();
408 std::fs::write(
409 dir.path().join("outcome.json"),
410 r#"{"schema_version":1,"outcome_id":"outcome-1","task_id":"other","accepted":"accepted","quality_score_milli":900,"signals":{"build":"passed","tests":"passed","lint":"passed","typecheck":"not_run","completion":"passed","pr":"not_run","correction":"not_run","rollback":"not_run","retry":"not_run"},"evidence_refs":[],"observed_at":"2026-08-01T09:01:00Z"}"#,
411 )
412 .unwrap();
413 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
414 assert!(!dir.path().join(RECORD_NAME).exists());
415 }
416
417 #[test]
418 fn incomplete_lineage_fails_without_record() {
419 let dir = fixture();
420 std::fs::remove_file(dir.path().join("context_plan.json")).unwrap();
421 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
422 assert!(!dir.path().join(RECORD_NAME).exists());
423 }
424
425 #[test]
426 fn malformed_typed_outcome_fails_without_record() {
427 let dir = fixture();
428 std::fs::write(
429 dir.path().join("outcome.json"),
430 r#"{"schema_version":1,"task_id":"task-1","outcome_id":"outcome-1"}"#,
431 )
432 .unwrap();
433 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
434 assert!(!dir.path().join(RECORD_NAME).exists());
435 }
436
437 #[test]
438 fn oversized_artifact_fails_without_record() {
439 let dir = fixture();
440 let file = std::fs::OpenOptions::new()
441 .write(true)
442 .open(dir.path().join("baseline.json"))
443 .unwrap();
444 file.set_len(MAX_ARTIFACT_BYTES + 1).unwrap();
445 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
446 assert!(!dir.path().join(RECORD_NAME).exists());
447 }
448
449 #[test]
450 fn conflicting_record_is_not_overwritten() {
451 let dir = fixture();
452 let path = dir.path().join(RECORD_NAME);
453 std::fs::write(&path, b"owned").unwrap();
454 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
455 assert_eq!(std::fs::read(path).unwrap(), b"owned");
456 }
457
458 #[cfg(unix)]
459 #[test]
460 fn symlinked_source_is_rejected() {
461 use std::os::unix::fs::symlink;
462 let dir = fixture();
463 let target = dir.path().join("real-task.json");
464 std::fs::rename(dir.path().join("task_envelope.json"), &target).unwrap();
465 symlink(&target, dir.path().join("task_envelope.json")).unwrap();
466 assert!(migrate_legacy_task_receipt(dir.path()).is_err());
467 assert!(!dir.path().join(RECORD_NAME).exists());
468 }
469}