use anyhow::{bail, Context, Result};
use chrono::{SecondsFormat, Utc};
use serde::{Deserialize, Serialize};
use serde_json::json;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0);
pub(crate) fn acquire_store_lock(store: &Path) -> Result<File> {
let metadata = store.join(".artifact-store");
fs::create_dir_all(&metadata)?;
let path = metadata.join("store.lock");
let file = OpenOptions::new()
.create(true)
.truncate(false)
.read(true)
.write(true)
.open(&path)
.with_context(|| format!("opening artifact store lock {}", path.display()))?;
file.lock()
.with_context(|| format!("locking artifact store {}", store.display()))?;
Ok(file)
}
pub(crate) fn write_atomic_unique(path: &Path, content: &[u8]) -> Result<()> {
let parent = path
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
.unwrap_or_else(|| Path::new("."));
fs::create_dir_all(parent)?;
let filename = path
.file_name()
.and_then(|value| value.to_str())
.unwrap_or("artifact");
let (temporary, mut file) = loop {
let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let candidate = parent.join(format!(
".{filename}.sdd-tmp-{}-{sequence}",
std::process::id()
));
match OpenOptions::new()
.create_new(true)
.write(true)
.open(&candidate)
{
Ok(file) => break (candidate, file),
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => {
return Err(error).with_context(|| {
format!("creating atomic temporary file {}", candidate.display())
});
}
}
};
let result = (|| -> Result<()> {
file.write_all(content)
.with_context(|| format!("writing atomic temporary {}", temporary.display()))?;
file.sync_all()
.with_context(|| format!("syncing atomic temporary {}", temporary.display()))?;
drop(file);
fs::rename(&temporary, path).with_context(|| {
format!(
"committing atomic write {} -> {}",
temporary.display(),
path.display()
)
})?;
Ok(())
})();
if result.is_err() {
let _ = fs::remove_file(&temporary);
}
result
}
#[derive(Clone, Debug)]
pub(crate) struct ArtifactLocator {
root: PathBuf,
}
impl ArtifactLocator {
pub(crate) fn new(root: &Path) -> Self {
Self {
root: root.to_path_buf(),
}
}
pub(crate) fn locate(&self, slug: &str, worktree_key: &str) -> Option<PathBuf> {
let direct = self.root.join("docs").join(slug);
let root_is_worktree = self.root.join(".git").is_file();
if root_is_worktree && direct.is_dir() {
return Some(direct);
}
let dedicated = self
.root
.join(".worktree")
.join(worktree_key)
.join("docs")
.join(slug);
if dedicated.is_dir() {
return Some(dedicated);
}
let discovered = self
.discover_paths()
.ok()?
.into_iter()
.find(|path| path.file_name().is_some_and(|name| name == slug));
discovered.or_else(|| direct.is_dir().then_some(direct))
}
pub(crate) fn target(&self, slug: &str, worktree_key: &str) -> PathBuf {
if let Some(existing) = self.locate(slug, worktree_key) {
return existing;
}
let worktree = self.root.join(".worktree").join(worktree_key);
if worktree.is_dir() {
worktree.join("docs").join(slug)
} else {
self.root.join("docs").join(slug)
}
}
pub(crate) fn discover_paths(&self) -> Result<Vec<PathBuf>> {
let mut stores = Vec::new();
collect_stores(&self.root.join("docs"), &mut stores)?;
let worktrees = self.root.join(".worktree");
if worktrees.is_dir() {
for entry in fs::read_dir(worktrees)? {
let docs = entry?.path().join("docs");
collect_stores(&docs, &mut stores)?;
}
}
stores.sort();
stores.dedup();
Ok(stores)
}
}
fn collect_stores(docs: &Path, stores: &mut Vec<PathBuf>) -> Result<()> {
if !docs.is_dir() {
return Ok(());
}
for entry in fs::read_dir(docs)? {
let path = entry?.path();
if path.is_dir() && path.join("traceability-map.yaml").is_file() {
stores.push(path);
}
}
Ok(())
}
#[derive(Clone, Debug)]
pub(crate) struct ArtifactStoreService {
project_root: PathBuf,
store: PathBuf,
orchestration: String,
}
#[derive(Clone, Debug)]
pub(crate) struct SaveRequest<'a> {
pub(crate) stage: &'a str,
pub(crate) filename: &'a str,
pub(crate) content: &'a [u8],
pub(crate) state: &'a str,
pub(crate) validation_status: &'a str,
pub(crate) if_match: Option<&'a str>,
}
#[derive(Clone, Debug, Serialize)]
pub(crate) struct SaveOutcome {
pub(crate) path: PathBuf,
pub(crate) revision: u64,
pub(crate) sha256: String,
pub(crate) size_bytes: u64,
pub(crate) updated_at: String,
pub(crate) validation_status: String,
pub(crate) superseded: bool,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub(crate) struct JournalRecord {
operation_id: String,
kind: String,
status: String,
stage: String,
target: String,
snapshot: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
index_target: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
index_snapshot: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
index_sha256: Option<String>,
revision: u64,
sha256: String,
previous_sha256: Option<String>,
timestamp: String,
}
#[derive(Clone, Debug, Serialize)]
pub(crate) struct DoctorReport {
pub(crate) status: &'static str,
pub(crate) store: PathBuf,
pub(crate) pending_transactions: usize,
pub(crate) corrupt_history_entries: Vec<String>,
pub(crate) warnings: Vec<String>,
}
#[derive(Clone, Debug, Serialize)]
pub(crate) struct RepairReport {
pub(crate) applied: bool,
pub(crate) recovered: Vec<String>,
pub(crate) already_consistent: Vec<String>,
pub(crate) unresolved: Vec<String>,
}
impl ArtifactStoreService {
pub(crate) fn new(project_root: &Path, store: &Path, orchestration: &str) -> Self {
Self {
project_root: project_root.to_path_buf(),
store: store.to_path_buf(),
orchestration: orchestration.to_string(),
}
}
pub(crate) fn save(&self, request: SaveRequest<'_>) -> Result<SaveOutcome> {
if request.state == "approved" && request.validation_status != "valid" {
bail!("approved artifact must pass schema validation");
}
fs::create_dir_all(&self.store)?;
validate_stage(request.stage)?;
let lock = acquire_store_lock(&self.store)?;
let pending = pending_records(&self.journal_records()?);
if !pending.is_empty() {
bail!(
"artifact store has {} pending transaction(s); run `sdd artifact repair` before saving",
pending.len()
);
}
let target = self.safe_store_path(request.filename)?;
let previous_sha256 = if target.is_file() {
Some(hash(&fs::read(&target)?))
} else {
None
};
if let Some(expected) = request.if_match {
if previous_sha256.as_deref() != Some(expected) {
bail!(
"artifact revision conflict: --if-match expected `{expected}`, current is `{}`",
previous_sha256.as_deref().unwrap_or("missing")
);
}
}
let revision = self.next_revision(request.stage)?;
let sha256 = hash(request.content);
let updated_at = timestamp();
let history = self
.history_dir()
.join(request.stage)
.join(format!("{revision}.md"));
let index_target = self.store.join("traceability-map.yaml");
let index_snapshot = self
.history_dir()
.join(request.stage)
.join(format!("{revision}.traceability-map.yaml"));
let index_content = self.updated_index(&request, revision, &sha256, &updated_at)?;
let index_sha256 = hash(&index_content);
let operation_id = format!("{}-{revision}-{}", request.stage, &sha256[..12]);
let mut record = JournalRecord {
operation_id,
kind: "save".to_string(),
status: "pending".to_string(),
stage: request.stage.to_string(),
target: relative(&self.store, &target),
snapshot: relative(&self.store, &history),
index_target: Some(relative(&self.store, &index_target)),
index_snapshot: Some(relative(&self.store, &index_snapshot)),
index_sha256: Some(index_sha256),
revision,
sha256: sha256.clone(),
previous_sha256,
timestamp: updated_at.clone(),
};
self.append_journal(&record)?;
write_atomic_unique(&history, request.content)?;
write_atomic_unique(&index_snapshot, &index_content)?;
write_atomic_unique(&target, request.content)?;
write_atomic_unique(&index_target, &index_content)?;
record.status = "committed".to_string();
record.timestamp = timestamp();
self.append_journal(&record)?;
self.cleanup_recovery_snapshots(&record);
let outcome = SaveOutcome {
path: target,
revision,
sha256,
size_bytes: request.content.len() as u64,
updated_at,
validation_status: request.validation_status.to_string(),
superseded: record.previous_sha256.is_some(),
};
drop(lock);
self.record_event("artifact_saved", request.stage, &outcome)?;
if outcome.superseded {
self.record_event("artifact_superseded", request.stage, &outcome)?;
}
Ok(outcome)
}
pub(crate) fn record_linked(&self, system: &str, url: &str, stage: Option<&str>) -> Result<()> {
crate::runtime::trace::record_event(
&self.project_root,
"events.jsonl",
json!({
"ts": timestamp(),
"kind": "artifact_linked",
"status": "recorded",
"orchestration": self.orchestration,
"stage": stage,
"target": relative(&self.project_root, &self.store.join("traceability-map.yaml")),
"system": system,
"url_sha256": hash(url.as_bytes()),
}),
)?;
Ok(())
}
pub(crate) fn history(&self) -> Result<Vec<JournalRecord>> {
Ok(self
.journal_records()?
.into_iter()
.filter(|record| record.kind == "save" && record.status == "committed")
.collect())
}
pub(crate) fn doctor(&self) -> Result<DoctorReport> {
let records = self.journal_records()?;
let pending = pending_records(&records);
let mut corrupt_history_entries = Vec::new();
for record in &pending {
let snapshot = self.safe_store_path(&record.snapshot)?;
match fs::read(&snapshot) {
Ok(bytes) if hash(&bytes) == record.sha256 => {}
Ok(_) => {
corrupt_history_entries.push(format!("{}: hash divergente", record.snapshot))
}
Err(_) => corrupt_history_entries.push(format!("{}: ausente", record.snapshot)),
}
match (
record.index_target.as_deref(),
record.index_snapshot.as_deref(),
record.index_sha256.as_deref(),
) {
(Some(target), Some(relative), Some(expected)) => {
self.safe_store_path(target)?;
let snapshot = self.safe_store_path(relative)?;
match fs::read(&snapshot) {
Ok(bytes) if hash(&bytes) == expected => {}
Ok(_) => corrupt_history_entries
.push(format!("{relative}: index snapshot hash divergente")),
Err(_) => corrupt_history_entries
.push(format!("{relative}: index snapshot ausente")),
}
}
(None, None, None) => {}
_ => corrupt_history_entries.push(format!(
"{}: metadados de index incompletos",
record.operation_id
)),
}
}
if let Ok(text) = fs::read_to_string(self.store.join("traceability-map.yaml")) {
if let Ok(document) = serde_yaml::from_str::<serde_yaml::Value>(&text) {
if let Some(artifacts) = document
.get("artifacts")
.and_then(serde_yaml::Value::as_mapping)
{
for artifact in artifacts.values() {
let Some(filename) =
artifact.get("file").and_then(serde_yaml::Value::as_str)
else {
continue;
};
let Some(expected) =
artifact.get("sha256").and_then(serde_yaml::Value::as_str)
else {
continue;
};
let target = self.safe_store_path(filename)?;
match fs::read(&target) {
Ok(bytes) if hash(&bytes) == expected => {}
Ok(_) => corrupt_history_entries
.push(format!("{filename}: target hash divergente")),
Err(_) => {
corrupt_history_entries.push(format!("{filename}: target ausente"))
}
}
}
}
}
}
let mut warnings = Vec::new();
if records
.iter()
.filter(|record| record.status == "committed")
.any(|record| {
self.safe_store_path(&record.snapshot)
.is_ok_and(|path| path.exists())
|| record.index_snapshot.as_deref().is_some_and(|snapshot| {
self.safe_store_path(snapshot)
.is_ok_and(|path| path.exists())
})
})
{
warnings.push(
"snapshots transitórios de transações já confirmadas; rode `sdd artifact repair --apply` para limpar"
.to_string(),
);
}
if !self.store.join("traceability-map.yaml").is_file() {
warnings.push("traceability-map.yaml ausente".to_string());
}
let status =
if corrupt_history_entries.is_empty() && pending.is_empty() && warnings.is_empty() {
"pass"
} else {
"warn"
};
Ok(DoctorReport {
status,
store: self.store.clone(),
pending_transactions: pending.len(),
corrupt_history_entries,
warnings,
})
}
pub(crate) fn repair(&self, apply: bool) -> Result<RepairReport> {
let _lock = apply.then(|| acquire_store_lock(&self.store)).transpose()?;
let records = self.journal_records()?;
let pending = pending_records(&records);
let mut report = RepairReport {
applied: apply,
recovered: Vec::new(),
already_consistent: Vec::new(),
unresolved: Vec::new(),
};
for mut record in pending {
let target = self.safe_store_path(&record.target)?;
let snapshot = self.safe_store_path(&record.snapshot)?;
let artifact_bytes = match fs::read(&snapshot) {
Ok(bytes) if hash(&bytes) == record.sha256 => bytes,
_ => {
report.unresolved.push(record.target.clone());
continue;
}
};
let index_plan = match (
record.index_target.as_deref(),
record.index_snapshot.as_deref(),
record.index_sha256.as_deref(),
) {
(Some(target), Some(snapshot), Some(expected)) => {
let target = self.safe_store_path(target)?;
let snapshot = self.safe_store_path(snapshot)?;
match fs::read(&snapshot) {
Ok(bytes) if hash(&bytes) == expected => Some((target, bytes, expected)),
_ => {
report.unresolved.push(record.target.clone());
continue;
}
}
}
(None, None, None) => None,
_ => {
report.unresolved.push(record.target.clone());
continue;
}
};
let target_matches = fs::read(&target).is_ok_and(|bytes| hash(&bytes) == record.sha256);
let index_matches = index_plan.as_ref().is_none_or(|(target, _, expected)| {
fs::read(target).is_ok_and(|bytes| hash(&bytes) == *expected)
});
if target_matches && index_matches {
report.already_consistent.push(record.target.clone());
} else {
report.recovered.push(record.target.clone());
if apply {
if !target_matches {
write_atomic_unique(&target, &artifact_bytes)?;
}
if let Some((index_target, index_bytes, _)) = &index_plan {
if !index_matches {
write_atomic_unique(index_target, index_bytes)?;
}
}
}
}
if apply {
record.status = "committed".to_string();
record.timestamp = timestamp();
self.append_journal(&record)?;
self.cleanup_recovery_snapshots(&record);
}
}
if apply {
for record in records.iter().filter(|record| record.status == "committed") {
self.cleanup_recovery_snapshots(record);
}
}
Ok(report)
}
fn updated_index(
&self,
request: &SaveRequest<'_>,
revision: u64,
sha256: &str,
updated_at: &str,
) -> Result<Vec<u8>> {
let path = self.store.join("traceability-map.yaml");
let mut document = if path.is_file() {
serde_yaml::from_str::<serde_yaml::Value>(&fs::read_to_string(&path)?)
.with_context(|| format!("parsing traceability map {}", path.display()))?
} else {
serde_yaml::Value::Mapping(serde_yaml::Mapping::new())
};
if document.is_null() {
document = serde_yaml::Value::Mapping(serde_yaml::Mapping::new());
}
let root = document
.as_mapping_mut()
.ok_or_else(|| anyhow::anyhow!("traceability-map root must be a mapping"))?;
let orchestration = ensure_mapping(root, "orchestration")?;
insert_if_missing(
orchestration,
"name",
serde_yaml::Value::String(self.orchestration.clone()),
);
orchestration.insert(
yaml_key("state"),
serde_yaml::Value::String(
if request.stage == "memory" {
"completed"
} else {
"in_progress"
}
.to_string(),
),
);
let artifacts = ensure_mapping(root, "artifacts")?;
let stage = ensure_mapping(artifacts, request.stage)?;
stage.insert(
yaml_key("file"),
serde_yaml::Value::String(request.filename.to_string()),
);
stage.insert(
yaml_key("state"),
serde_yaml::Value::String(request.state.to_string()),
);
stage.insert(
yaml_key("approved_at"),
serde_yaml::Value::String(
if request.state == "approved" {
updated_at
} else {
""
}
.to_string(),
),
);
stage.insert(
yaml_key("revision"),
serde_yaml::Value::Number(revision.into()),
);
stage.insert(
yaml_key("sha256"),
serde_yaml::Value::String(sha256.to_string()),
);
stage.insert(
yaml_key("size_bytes"),
serde_yaml::Value::Number((request.content.len() as u64).into()),
);
stage.insert(
yaml_key("updated_at"),
serde_yaml::Value::String(updated_at.to_string()),
);
stage.insert(
yaml_key("validation_status"),
serde_yaml::Value::String(request.validation_status.to_string()),
);
Ok(serde_yaml::to_string(&document)?.into_bytes())
}
fn record_event(&self, kind: &str, stage: &str, outcome: &SaveOutcome) -> Result<()> {
crate::runtime::trace::record_event(
&self.project_root,
"events.jsonl",
json!({
"ts": outcome.updated_at,
"kind": kind,
"status": "recorded",
"orchestration": self.orchestration,
"stage": stage,
"target": relative(&self.project_root, &outcome.path),
"revision": outcome.revision,
"sha256": outcome.sha256,
"size_bytes": outcome.size_bytes,
"validation_status": outcome.validation_status,
}),
)?;
Ok(())
}
fn next_revision(&self, stage: &str) -> Result<u64> {
let mut max_revision = self
.history()?
.into_iter()
.filter(|record| record.stage == stage)
.map(|record| record.revision)
.max()
.unwrap_or(0);
if let Ok(text) = fs::read_to_string(self.store.join("traceability-map.yaml")) {
if let Ok(document) = serde_yaml::from_str::<serde_yaml::Value>(&text) {
let indexed = document
.get("artifacts")
.and_then(|artifacts| artifacts.get(stage))
.and_then(|artifact| artifact.get("revision"))
.and_then(serde_yaml::Value::as_u64)
.unwrap_or(0);
max_revision = max_revision.max(indexed);
}
}
Ok(max_revision + 1)
}
fn cleanup_recovery_snapshots(&self, record: &JournalRecord) {
let mut snapshots = vec![record.snapshot.as_str()];
if let Some(snapshot) = record.index_snapshot.as_deref() {
snapshots.push(snapshot);
}
for snapshot in snapshots {
if let Ok(path) = self.safe_store_path(snapshot) {
let _ = fs::remove_file(&path);
if let Some(parent) = path.parent() {
let _ = fs::remove_dir(parent);
}
}
}
}
fn metadata_dir(&self) -> PathBuf {
self.store.join(".artifact-store")
}
fn history_dir(&self) -> PathBuf {
self.metadata_dir().join("history")
}
fn journal_path(&self) -> PathBuf {
self.metadata_dir().join("journal.jsonl")
}
fn append_journal(&self, record: &JournalRecord) -> Result<()> {
let path = self.journal_path();
let mut existing = if path.is_file() {
fs::read_to_string(&path)?
} else {
String::new()
};
existing.push_str(&serde_json::to_string(record)?);
existing.push('\n');
write_atomic_unique(&path, existing.as_bytes())
}
fn journal_records(&self) -> Result<Vec<JournalRecord>> {
let path = self.journal_path();
if !path.is_file() {
return Ok(Vec::new());
}
fs::read_to_string(&path)?
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| {
serde_json::from_str(line)
.with_context(|| format!("journal inválido em {}", path.display()))
})
.collect()
}
fn safe_store_path(&self, relative: &str) -> Result<PathBuf> {
let path = Path::new(relative);
if relative.is_empty()
|| path.is_absolute()
|| path.components().any(|component| {
matches!(
component,
Component::CurDir
| Component::ParentDir
| Component::RootDir
| Component::Prefix(_)
)
})
{
bail!("unsafe journal path outside artifact store: {relative}");
}
Ok(self.store.join(path))
}
}
fn validate_stage(stage: &str) -> Result<()> {
let mut components = Path::new(stage).components();
if stage.is_empty()
|| !matches!(components.next(), Some(Component::Normal(_)))
|| components.next().is_some()
{
bail!("unsafe artifact stage: {stage}");
}
Ok(())
}
fn yaml_key(key: &str) -> serde_yaml::Value {
serde_yaml::Value::String(key.to_string())
}
fn ensure_mapping<'a>(
parent: &'a mut serde_yaml::Mapping,
key: &str,
) -> Result<&'a mut serde_yaml::Mapping> {
let key = yaml_key(key);
if !parent.contains_key(&key) {
parent.insert(
key.clone(),
serde_yaml::Value::Mapping(serde_yaml::Mapping::new()),
);
}
parent
.get_mut(&key)
.and_then(serde_yaml::Value::as_mapping_mut)
.ok_or_else(|| anyhow::anyhow!("traceability-map field must be a mapping: {key:?}"))
}
fn insert_if_missing(mapping: &mut serde_yaml::Mapping, key: &str, value: serde_yaml::Value) {
let key = yaml_key(key);
if !mapping.contains_key(&key) {
mapping.insert(key, value);
}
}
fn pending_records(records: &[JournalRecord]) -> Vec<JournalRecord> {
let mut pending = BTreeMap::new();
let mut committed = BTreeSet::new();
for record in records {
if record.status == "pending" {
pending.insert(record.operation_id.clone(), record.clone());
} else if record.status == "committed" {
committed.insert(record.operation_id.clone());
}
}
pending
.into_iter()
.filter_map(|(id, record)| (!committed.contains(&id)).then_some(record))
.collect()
}
fn hash(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
format!("{:x}", hasher.finalize())
}
fn timestamp() -> String {
Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true)
}
fn relative(root: &Path, path: &Path) -> String {
path.strip_prefix(root)
.unwrap_or(path)
.to_string_lossy()
.replace('\\', "/")
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Barrier};
use std::thread;
use tempfile::tempdir;
#[test]
fn repair_rejects_journal_paths_that_escape_the_store() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
let metadata = store.join(".artifact-store");
fs::create_dir_all(metadata.join("history/idea")).unwrap();
fs::write(metadata.join("history/idea/1.md"), "safe snapshot").unwrap();
let record = json!({
"operation_id": "malicious",
"kind": "save",
"status": "pending",
"stage": "idea",
"target": "../../escaped.md",
"snapshot": ".artifact-store/history/idea/1.md",
"revision": 1,
"sha256": hash(b"safe snapshot"),
"previous_sha256": null,
"timestamp": timestamp(),
});
fs::write(
metadata.join("journal.jsonl"),
format!("{}\n", serde_json::to_string(&record).unwrap()),
)
.unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let error = service.repair(true).unwrap_err().to_string();
assert!(error.contains("unsafe journal path"), "{error}");
assert!(!tmp.path().join("escaped.md").exists());
}
#[test]
fn repair_rejects_corrupt_snapshot_without_committing_it() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
let metadata = store.join(".artifact-store");
let snapshot = metadata.join("history/idea/1.md");
fs::create_dir_all(snapshot.parent().unwrap()).unwrap();
fs::write(&snapshot, "corrupt snapshot").unwrap();
let record = json!({
"operation_id": "corrupt",
"kind": "save",
"status": "pending",
"stage": "idea",
"target": "01-idea.md",
"snapshot": ".artifact-store/history/idea/1.md",
"revision": 1,
"sha256": hash(b"expected snapshot"),
"previous_sha256": null,
"timestamp": timestamp(),
});
fs::write(
metadata.join("journal.jsonl"),
format!("{}\n", serde_json::to_string(&record).unwrap()),
)
.unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let report = service.repair(true).unwrap();
assert_eq!(report.unresolved, vec!["01-idea.md"]);
assert!(report.recovered.is_empty());
assert!(!store.join("01-idea.md").exists());
assert_eq!(service.doctor().unwrap().pending_transactions, 1);
}
#[test]
fn save_upgrades_legacy_map_without_stage_and_commits_metadata() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
fs::create_dir_all(&store).unwrap();
fs::write(
store.join("traceability-map.yaml"),
"orchestration:\n name: flow\n state: draft\nartifacts: {}\n",
)
.unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let outcome = service
.save(SaveRequest {
stage: "idea",
filename: "01-idea.md",
content: b"legacy compatible",
state: "approved",
validation_status: "valid",
if_match: None,
})
.unwrap();
let map: serde_yaml::Value =
serde_yaml::from_str(&fs::read_to_string(store.join("traceability-map.yaml")).unwrap())
.unwrap();
assert_eq!(map["artifacts"]["idea"]["file"], "01-idea.md");
assert_eq!(map["artifacts"]["idea"]["state"], "approved");
assert_eq!(map["artifacts"]["idea"]["revision"], outcome.revision);
assert_eq!(map["artifacts"]["idea"]["sha256"], outcome.sha256);
assert_eq!(map["orchestration"]["state"], "in_progress");
}
#[test]
fn save_does_not_write_artifact_or_event_when_map_cannot_be_committed() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
fs::create_dir_all(&store).unwrap();
fs::write(store.join("traceability-map.yaml"), "artifacts: [\n").unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
assert!(service
.save(SaveRequest {
stage: "idea",
filename: "01-idea.md",
content: b"must not commit",
state: "recorded",
validation_status: "valid",
if_match: None,
})
.is_err());
assert!(!store.join("01-idea.md").exists());
assert!(!tmp.path().join(".sdd/events.jsonl").exists());
}
#[test]
fn repair_rolls_forward_artifact_and_map_as_one_pending_operation() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
let metadata = store.join(".artifact-store");
let history = metadata.join("history/idea");
fs::create_dir_all(&history).unwrap();
let artifact = b"recovered artifact";
let map = b"orchestration:\n state: in_progress\nartifacts:\n idea:\n file: 01-idea.md\n revision: 1\n";
fs::write(history.join("1.md"), artifact).unwrap();
fs::write(history.join("1.traceability-map.yaml"), map).unwrap();
fs::write(
store.join("traceability-map.yaml"),
"orchestration:\n state: draft\nartifacts: {}\n",
)
.unwrap();
let record = json!({
"operation_id": "recover-both",
"kind": "save",
"status": "pending",
"stage": "idea",
"target": "01-idea.md",
"snapshot": ".artifact-store/history/idea/1.md",
"index_target": "traceability-map.yaml",
"index_snapshot": ".artifact-store/history/idea/1.traceability-map.yaml",
"index_sha256": hash(map),
"revision": 1,
"sha256": hash(artifact),
"previous_sha256": null,
"timestamp": timestamp(),
});
fs::write(
metadata.join("journal.jsonl"),
format!("{}\n", serde_json::to_string(&record).unwrap()),
)
.unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let report = service.repair(true).unwrap();
assert_eq!(report.recovered, vec!["01-idea.md"]);
assert_eq!(fs::read(store.join("01-idea.md")).unwrap(), artifact);
assert_eq!(fs::read(store.join("traceability-map.yaml")).unwrap(), map);
assert_eq!(service.doctor().unwrap().pending_transactions, 0);
}
#[test]
fn doctor_rejects_traceability_paths_outside_store() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
fs::create_dir_all(&store).unwrap();
let outside = tmp.path().join("secret.md");
fs::write(&outside, "secret").unwrap();
fs::write(
store.join("traceability-map.yaml"),
format!(
"artifacts:\n idea:\n file: ../../secret.md\n sha256: {}\n",
hash(b"secret")
),
)
.unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let error = service.doctor().unwrap_err().to_string();
assert!(error.contains("unsafe"), "{error}");
}
#[test]
fn save_ignores_stale_deterministic_temp_file_name() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
let history = store.join(".artifact-store/history/idea");
fs::create_dir_all(history.join(".1.md.sdd-tmp")).unwrap();
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
service
.save(SaveRequest {
stage: "idea",
filename: "01-idea.md",
content: b"unique temporary",
state: "recorded",
validation_status: "valid",
if_match: None,
})
.unwrap();
assert_eq!(
fs::read(store.join("01-idea.md")).unwrap(),
b"unique temporary"
);
}
#[test]
fn concurrent_if_match_allows_exactly_one_writer() {
let tmp = tempdir().unwrap();
let store = tmp.path().join("docs/flow");
let service = ArtifactStoreService::new(tmp.path(), &store, "flow");
let initial = service
.save(SaveRequest {
stage: "idea",
filename: "01-idea.md",
content: b"initial",
state: "recorded",
validation_status: "valid",
if_match: None,
})
.unwrap();
let expected = Arc::new(initial.sha256);
let barrier = Arc::new(Barrier::new(8));
let mut workers = Vec::new();
for index in 0..8 {
let service = service.clone();
let expected = Arc::clone(&expected);
let barrier = Arc::clone(&barrier);
workers.push(thread::spawn(move || {
let content = vec![b'0' + index; 512 * 1024];
barrier.wait();
service.save(SaveRequest {
stage: "idea",
filename: "01-idea.md",
content: &content,
state: "recorded",
validation_status: "valid",
if_match: Some(expected.as_str()),
})
}));
}
let results = workers
.into_iter()
.map(|worker| worker.join().unwrap())
.collect::<Vec<_>>();
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
for error in results.into_iter().filter_map(Result::err) {
assert!(error.to_string().contains("revision conflict"), "{error:#}");
}
assert_eq!(service.history().unwrap().len(), 2);
assert_eq!(service.doctor().unwrap().status, "pass");
}
}