{{ input.key }}
-
+
+
+ 来源：{{ input.source }}（可修改）
+
@@ -20443,6 +20446,16 @@ compare: [[flesh, peel]]`;
if (!response.ok) throw new Error(await extensionApiError(response, '读取拓展分析说明失败'));
return await response.json();
};
+ const loadUpstreamCandidates = async (extensionId) => {
+ const jobId = String(jobInfo.job_id || '').trim();
+ if (!jobId || jobId === 'default_job') return {};
+ try {
+ const response = await extensionApiFetch(`/api/lightweight-extension-stages/${encodeURIComponent(extensionId)}/upstream-candidates?job_id=${encodeURIComponent(jobId)}`);
+ if (!response.ok) return {};
+ const data = await response.json();
+ return (data && data.candidates) || {};
+ } catch (_) { return {}; }
+ };
const isLightweightExtensionId = (extensionId) => (lightweightExtensions.value || []).some(item => item.extension_id === extensionId);
const lightweightRunById = (runs, runId) => (runs || []).find(item => item && item.run_id === runId) || {};
const lightweightResumeNext = (action) => ({
@@ -20880,11 +20893,13 @@ compare: [[flesh, peel]]`;
});
if (hitIndex >= 0) {
input.value = remaining.splice(hitIndex, 1)[0];
+ input.source = '';
}
}
const emptyInputs = lightweightNlRun.inputs.filter(item => !String(item.value || '').trim());
for (let i = 0; i < remaining.length && i < emptyInputs.length; i++) {
emptyInputs[i].value = remaining[i];
+ emptyInputs[i].source = '';
}
const matched = uploaded.length - remaining.length;
ElMessage.success(`已上传 ${uploaded.length} 个文件${matched ? `，已自动匹配 ${matched} 个输入槽` : ''}`);
@@ -20911,6 +20926,7 @@ compare: [[flesh, peel]]`;
const targetInput = lightweightNlRun.inputs.find(item => !String(item.value || '').trim()) || lightweightNlRun.inputs[0];
if (targetInput) {
targetInput.value = Object.values(imported)[0] || '';
+ targetInput.source = '';
}
ElMessage.success(`已导入 ${Object.keys(imported).length} 个路径`);
} catch (err) {
@@ -20923,7 +20939,7 @@ compare: [[flesh, peel]]`;
if (!extensionId) return null;
try {
const help = await lightweightHelp(extensionId);
- const inputs = (help.inputs || []).filter(item => item && item.replaceable).map(item => ({ key: item.path, value: '' }));
+ const inputs = (help.inputs || []).filter(item => item && item.replaceable).map(item => ({ key: item.path, value: '', source: '' }));
const parameters = (help.parameters || []).map(item => ({ key: item.fact_id, flag: item.flag || item.fact_id, display_name: item.display_name || '', value: item.value || '' }));
const slots = (intentParams && intentParams.llm_slots) || {};
const slotInputs = slots.input_overrides || {};
@@ -20932,6 +20948,15 @@ compare: [[flesh, peel]]`;
const value = slotInputs[item.key];
if (value) item.value = value;
});
+ const upstreamCandidates = await loadUpstreamCandidates(extensionId);
+ inputs.forEach(item => {
+ if (String(item.value || '').trim()) return;
+ const candidate = upstreamCandidates[item.key];
+ if (candidate && candidate.path) {
+ item.value = candidate.path;
+ item.source = `上游样本 ${candidate.source_extension_id}/${candidate.run_id}`;
+ }
+ });
parameters.forEach(item => {
const value = slotParameters[item.key];
if (value) item.value = value;
diff --git a/gfa_core/extension_stage_lightweight/api.py b/gfa_core/extension_stage_lightweight/api.py
index 11f94800e..9e22299ae 100644
--- a/gfa_core/extension_stage_lightweight/api.py
+++ b/gfa_core/extension_stage_lightweight/api.py
@@ -20,6 +20,7 @@ from pydantic import BaseModel, Field
from gfa_core.extension_stage.raw_package_normalizer import normalize_raw_package
from .service import LightweightExtensionService
+from . import upstream
UPLOAD_CHUNK_BYTES = 1024 * 1024
@@ -655,6 +656,54 @@ def create_lightweight_extension_router(
return service_factory(settings).help(_safe_component(extension_id, name="extension_id"))
except ValueError as exc:
raise _http_error(404, "help", exc) from exc
+ @router.get("/{extension_id}/upstream-candidates")
+ async def upstream_candidates(extension_id: str, job_id: str = "") -> dict[str, Any]:
+ """同 job 内其他轻量拓展成功运行产物对契约输入槽的候选（供运行向导预填）。"""
+ try:
+ extension_id = _safe_component(extension_id, name="extension_id")
+ help_payload = service_factory(settings).help(extension_id)
+ except ValueError as exc:
+ raise _http_error(404, "upstream-candidates", exc) from exc
+ job_id = str(job_id or "").strip()
+ base = {"extension_id": extension_id, "job_id": job_id, "candidates": {}, "warnings": []}
+ if not job_id or job_id in {".", ".."} or "/" in job_id or "\\" in job_id:
+ return base
+ try:
+ record = service_factory(settings).get(extension_id)
+ contract = record.get("runtime_binding_contract") or {}
+ patchable_paths = {
+ str(target.get("binding_key") or "")[len("path:") :]
+ for target in contract.get("patch_targets") or []
+ if isinstance(target, dict)
+ and str(target.get("binding_key") or "").startswith("path:")
+ }
+ selected = upstream.resolve_upstream_slots(
+ settings.workspace_root,
+ job_id,
+ help_payload.get("inputs") or [],
+ exclude_extension_id=extension_id,
+ patchable_paths=patchable_paths,
+ )
+ job_root = settings.workspace_root / job_id
+ candidates: dict[str, Any] = {}
+ for slot, candidate in selected.items():
+ try:
+ relative = candidate.path.relative_to(job_root).as_posix()
+ except ValueError:
+ try:
+ relative = candidate.path.resolve().relative_to(job_root.resolve()).as_posix()
+ except ValueError:
+ continue
+ candidates[slot] = {
+ "path": f"workspace/{job_id}/{relative}",
+ "source_extension_id": candidate.source_extension_id,
+ "run_id": candidate.run_id,
+ "is_dir": candidate.is_dir,
+ }
+ base["candidates"] = candidates
+ return base
+ except OSError:
+ return base
@router.post("/intent")
async def intent(req: _IntentRequest) -> dict[str, Any]:
diff --git a/gfa_core/extension_stage_lightweight/nl_intent.py b/gfa_core/extension_stage_lightweight/nl_intent.py
index 6ccbac96d..b4cd1b0e3 100644
--- a/gfa_core/extension_stage_lightweight/nl_intent.py
+++ b/gfa_core/extension_stage_lightweight/nl_intent.py
@@ -40,7 +40,7 @@ _RESPONSE_SYSTEM_PROMPT = (
"只能引用该拓展 parameters[].fact_id。\n"
"3. 参数值只能来自用户消息原文；用户没提到的参数不要出现。\n"
"4. 用户提到的文件优先与 job_files 匹配，统一写成 \"input//<文件名>\" 或 "
- "\"workspace//<文件名>\" 的 job 相对形式；禁止绝对路径。\n"
+ "\"workspace//<文件名>\" 的 job 相对形式；禁止绝对路径。用户未指明具体文件时，可引用 workspace//extension_runs/<其他拓展>/runtime_*/outputs/ 下已有的上游产物。\n"
"5. 禁止编造任何目录、路径、参数或数值。\n"
'6. 输出必须能被 JSON.parse 直接解析。'
)
diff --git a/gfa_core/extension_stage_lightweight/runtime_binding.py b/gfa_core/extension_stage_lightweight/runtime_binding.py
index 135a6d240..7e010fc7b 100644
--- a/gfa_core/extension_stage_lightweight/runtime_binding.py
+++ b/gfa_core/extension_stage_lightweight/runtime_binding.py
@@ -53,6 +53,24 @@ def _sha256_file(path: Path) -> str:
return digest.hexdigest()
+def _sha256_tree(source: Path) -> str:
+ """确定性目录摘要：每个普通文件的 (相对路径, 内容摘要) 级联；跳过符号链接本身。"""
+ digest = hashlib.sha256()
+ for path in sorted(source.rglob("*")):
+ relative = path.relative_to(source).as_posix()
+ if path.is_symlink():
+ continue
+ if path.is_file():
+ digest.update(relative.encode("utf-8"))
+ digest.update(b"\0")
+ digest.update(_sha256_file(path).encode("utf-8"))
+ digest.update(b"\0")
+ elif path.is_dir():
+ digest.update(relative.encode("utf-8"))
+ digest.update(b"\0dir\0")
+ return digest.hexdigest()
+
+
def _verify_raw_code(
inventory: dict[str, dict[str, Any]], raw_code: Path
) -> None:
@@ -631,10 +649,10 @@ def _resolve_inputs(
"message": f"Input has conflicting runtime bindings: {original_path}",
}
runtime_path = paths[0]
- if not runtime_path.is_file():
+ if not (runtime_path.is_file() or runtime_path.is_dir()):
return None, {
"code": "INPUT_PATH_NOT_FOUND",
- "message": f"Runtime input file does not exist: {runtime_path}",
+ "message": f"Runtime input file or directory does not exist: {runtime_path}",
}
bindings.append(
{
@@ -1765,7 +1783,7 @@ def _input_materialization_plan(
original = str(binding.get("original_path") or "")
lexical_source = Path(str(binding.get("runtime_path") or ""))
source = lexical_source.resolve()
- if not source.is_file():
+ if not (source.is_file() or source.is_dir()):
raise RuntimeBindingError(
"INPUT_PATH_NOT_FOUND", f"runtime input does not exist: {source}"
)
@@ -1782,15 +1800,26 @@ def _input_materialization_plan(
f"runtime input basename collides in work directory: {destination}",
)
used.add(destination)
- plan.append(
- {
- "original_path": original,
- "runtime_path": str(lexical_source),
- "materialized_path": destination,
- "method": "copy" if original in modified_originals else "symlink",
- "sha256": _sha256_file(source),
- }
- )
+ if source.is_dir():
+ plan.append(
+ {
+ "original_path": original,
+ "runtime_path": str(lexical_source),
+ "materialized_path": destination,
+ "method": "copy_tree",
+ "sha256": _sha256_tree(source),
+ }
+ )
+ else:
+ plan.append(
+ {
+ "original_path": original,
+ "runtime_path": str(lexical_source),
+ "materialized_path": destination,
+ "method": "copy" if original in modified_originals else "symlink",
+ "sha256": _sha256_file(source),
+ }
+ )
return plan
@@ -1927,7 +1956,9 @@ def render_runtime_package(
for item in materialization:
destination = work_dir / item["materialized_path"]
source = Path(item["runtime_path"])
- if item["method"] == "copy":
+ if item["method"] == "copy_tree":
+ shutil.copytree(source, destination, symlinks=True)
+ elif item["method"] == "copy":
shutil.copy2(source, destination)
else:
destination.symlink_to(source)
@@ -2228,6 +2259,20 @@ def _verify_runtime_inputs(
item.get("materialized_path"), field="materialized input path"
)
expected = item.get("sha256")
+ if item.get("method") == "copy_tree":
+ if not source.is_dir() or _sha256_tree(source) != expected:
+ raise RuntimeBindingError(
+ "INPUT_CONTENT_CHANGED",
+ f"runtime input directory content changed: {source}",
+ )
+ if after:
+ continue
+ if not materialized.is_dir() or _sha256_tree(materialized) != expected:
+ raise RuntimeBindingError(
+ "INPUT_CONTENT_CHANGED",
+ f"materialized input directory content changed: {materialized}",
+ )
+ continue
if not source.is_file() or _sha256_file(source) != expected:
raise RuntimeBindingError(
"INPUT_CONTENT_CHANGED",
@@ -2252,6 +2297,20 @@ def _isolate_runtime_inputs(
item.get("materialized_path"), field="materialized input path"
)
destination = work_dir / relative
+ if item.get("method") == "copy_tree":
+ if not destination.is_dir() or destination.is_symlink():
+ raise RuntimeBindingError(
+ "INPUT_ISOLATION_FAILED",
+ f"input directory is not an isolated copy: {relative}",
+ )
+ execution_inputs.append(
+ {
+ "materialized_path": relative,
+ "mode": "isolated_copy",
+ "source_path": str(source),
+ }
+ )
+ continue
if item.get("method") == "symlink":
file_descriptor, temporary_name = tempfile.mkstemp(
prefix=".bbav5-input-", dir=destination.parent
diff --git a/gfa_core/extension_stage_lightweight/service.py b/gfa_core/extension_stage_lightweight/service.py
index bddd03b85..715e81418 100644
--- a/gfa_core/extension_stage_lightweight/service.py
+++ b/gfa_core/extension_stage_lightweight/service.py
@@ -622,8 +622,10 @@ class LightweightExtensionService:
candidate = Path(os.path.abspath(Path(path).expanduser())).resolve()
resolved_roots = tuple(root.resolve() for root in roots)
inside = any(candidate.is_relative_to(root) for root in resolved_roots)
- if not candidate.is_file():
- raise ValueError(f"runtime input file not found: {candidate}")
+ if not candidate.exists():
+ raise ValueError(f"runtime input file or directory not found: {candidate}")
+ if not (candidate.is_file() or candidate.is_dir()):
+ raise ValueError(f"runtime input must be a file or directory: {candidate}")
if not inside:
raise ValueError(
f"runtime input must be a file inside the selected job: {candidate}"
diff --git a/gfa_core/extension_stage_lightweight/upstream.py b/gfa_core/extension_stage_lightweight/upstream.py
new file mode 100644
index 000000000..5d99ec8a1
--- /dev/null
+++ b/gfa_core/extension_stage_lightweight/upstream.py
@@ -0,0 +1,233 @@
+"""轻量版拓展分析的"上游样本继承"发现。
+
+从 workspace//extension_runs//runtime_*/outputs/ 枚举同 job 内其他轻量拓展的
+成功运行产物，为运行向导的空输入槽提供保守的候选填充。
+
+- 只读、不抛异常：任何解析失败都降级为无候选；
+- 只认轻量拓展的 runtime_* 运行目录（标准拓展的 run_* 无 runtime_execution_result.json，
+ 天然被成功判定排除）；
+- 匹配保守：basename 精确匹配优先，role 关键词必须唯一命中才采纳，否则留空让用户手填。
+
+对外入口：resolve_upstream_slots(workspace_root, job_id, contract_inputs, ...)。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+import re
+from pathlib import Path
+from typing import Any, Mapping
+
+MAX_UPSTREAM_RUNS = 100
+
+_DIRECTORY_HINT_TOKENS = ("directory", " dir", "folder", "bundle")
+
+_ROLE_STOPWORDS = {
+ "sample", "samples", "manifest", "with", "file", "files", "path", "paths",
+ "tab", "separated", "containing", "accession", "accessions", "genome",
+ "gff3", "protein", "cds", "peptide", "table", "list", "lists", "data",
+ "input", "output", "analysis", "columns", "column", "including", "that",
+ "from", "for", "and", "the",
+}
+
+
+@dataclass(frozen=True)
+class UpstreamCandidate:
+ path: Path # 词法路径（workspace//extension_runs/... 下，不 resolve）
+ relative: str # 相对 outputs 目录的 posix 路径
+ source_extension_id: str
+ run_id: str
+ is_dir: bool
+ run_time: float # run 目录 mtime，排序键
+
+
+def _run_result(run_dir: Path) -> dict[str, Any] | None:
+ try:
+ path = run_dir / "runtime_execution_result.json"
+ if not path.is_file():
+ return None
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, ValueError, TypeError):
+ return None
+ return data if isinstance(data, dict) else None
+
+
+def _run_success(run_dir: Path) -> bool:
+ result = _run_result(run_dir)
+ return bool(result) and result.get("success") is True
+
+
+def _run_time(run_dir: Path) -> float:
+ try:
+ return run_dir.stat().st_mtime
+ except OSError:
+ return 0.0
+
+
+def collect_job_upstream_outputs(
+ workspace_root: Path, job_id: str, *, exclude_extension_id: str = ""
+) -> list[UpstreamCandidate]:
+ """枚举同 job 下其他轻量拓展的成功运行产物（outputs/**）。任何失败返回空列表。"""
+ try:
+ extension_runs = Path(workspace_root) / job_id / "extension_runs"
+ if not extension_runs.is_dir():
+ return []
+ pool: list[UpstreamCandidate] = []
+ for ext_dir in sorted(extension_runs.iterdir()):
+ if not ext_dir.is_dir() or ext_dir.name == exclude_extension_id:
+ continue
+ for run_dir in sorted(ext_dir.glob("runtime_*")):
+ if len(pool) >= MAX_UPSTREAM_RUNS:
+ return pool
+ if not run_dir.is_dir() or not _run_success(run_dir):
+ continue
+ outputs = run_dir / "outputs"
+ if not outputs.is_dir():
+ continue
+ run_time = _run_time(run_dir)
+ for path in outputs.rglob("*"):
+ try:
+ if path.is_symlink():
+ continue
+ if path.is_file() or path.is_dir():
+ pool.append(
+ UpstreamCandidate(
+ path=path,
+ relative=path.relative_to(outputs).as_posix(),
+ source_extension_id=ext_dir.name,
+ run_id=run_dir.name,
+ is_dir=path.is_dir(),
+ run_time=run_time,
+ )
+ )
+ except OSError:
+ continue
+ return pool
+ except OSError:
+ return []
+
+
+def _slot_directory_like(slot: Mapping[str, Any]) -> bool:
+ text = str(slot.get("path") or "")
+ if text.endswith("/"):
+ return True
+ for field in ("role", "description"):
+ value = str(slot.get(field) or "").lower()
+ if any(token in value for token in _DIRECTORY_HINT_TOKENS):
+ return True
+ return False
+
+
+def _role_tokens(role: str) -> list[str]:
+ return [
+ token
+ for token in re.findall(r"[a-zA-Z0-9]{4,}", role)
+ if token not in _ROLE_STOPWORDS
+ ]
+
+
+def _match_slot(
+ slot: Mapping[str, Any],
+ pool: list[UpstreamCandidate],
+ patchable_paths: set[str] | None,
+) -> list[UpstreamCandidate]:
+ """保守匹配一个输入槽：命中即停，匹配不到返回空列表（不猜、不报错）。
+
+ 目录候选只进目录槽（_slot_directory_like 判定）；文件槽只认文件候选。
+ 精确同名匹配永远安全（不改名就不触发脚本替换），始终生效；
+ role 模糊匹配会把上游产物改成槽位名，只对可补丁（patch_targets 里声明过
+ path:<槽位path>）的槽位启用，否则留空让用户手填同名文件。
+ """
+ name = Path(str(slot.get("path") or "")).name.lower()
+ if not name:
+ return []
+ wants_dir = _slot_directory_like(slot)
+ typed = [candidate for candidate in pool if candidate.is_dir is wants_dir]
+ exact = [candidate for candidate in typed if candidate.path.name.lower() == name]
+ if exact:
+ return exact
+ if patchable_paths is not None and str(slot.get("path") or "") not in patchable_paths:
+ return []
+ tokens = _role_tokens(str(slot.get("role") or ""))
+ if not tokens:
+ return []
+ hits = typed
+ for token in tokens:
+ hits = [candidate for candidate in hits if token in candidate.path.name.lower()]
+ return hits
+
+
+def _candidate_rank(candidate: UpstreamCandidate) -> tuple[float, int, str]:
+ return (-candidate.run_time, len(Path(candidate.relative).parts), candidate.relative.lower())
+
+
+def find_upstream_candidates(
+ workspace_root: Path,
+ job_id: str,
+ contract_inputs: list[Mapping[str, Any]],
+ *,
+ exclude_extension_id: str = "",
+ patchable_paths: set[str] | None = None,
+) -> dict[str, list[UpstreamCandidate]]:
+ """为每个可替换输入槽收集上游候选（key = 契约 path）。
+
+ patchable_paths 为契约 patch_targets 中 path:<槽位path> 的集合；传入时
+ role 模糊匹配只对可补丁槽位启用（见 _match_slot）。
+ """
+ pool = collect_job_upstream_outputs(
+ workspace_root, job_id, exclude_extension_id=exclude_extension_id
+ )
+ result: dict[str, list[UpstreamCandidate]] = {}
+ for item in contract_inputs:
+ if not isinstance(item, Mapping) or not item.get("replaceable"):
+ continue
+ path = str(item.get("path") or "")
+ if not path:
+ continue
+ matched = _match_slot(item, pool, patchable_paths)
+ if matched:
+ result[path] = matched
+ return result
+
+
+def select_upstream_for_slots(
+ candidates: dict[str, list[UpstreamCandidate]],
+ contract_inputs: list[Mapping[str, Any]],
+) -> dict[str, UpstreamCandidate]:
+ """每槽收敛一个候选：最新 run 优先、outputs 相对路径最短；每个候选只能被一个槽占用。"""
+ selected: dict[str, UpstreamCandidate] = {}
+ used: set[Path] = set()
+ for item in contract_inputs:
+ if not isinstance(item, Mapping) or not item.get("replaceable"):
+ continue
+ path = str(item.get("path") or "")
+ if not path or path in selected:
+ continue
+ for candidate in sorted(candidates.get(path) or [], key=_candidate_rank):
+ if candidate.path in used:
+ continue
+ selected[path] = candidate
+ used.add(candidate.path)
+ break
+ return selected
+
+
+def resolve_upstream_slots(
+ workspace_root: Path,
+ job_id: str,
+ contract_inputs: list[Mapping[str, Any]],
+ *,
+ exclude_extension_id: str = "",
+ patchable_paths: set[str] | None = None,
+) -> dict[str, UpstreamCandidate]:
+ """一步：枚举上游成功产物 → 槽位保守匹配 → 收敛为每槽一个候选。"""
+ return select_upstream_for_slots(
+ find_upstream_candidates(
+ workspace_root,
+ job_id,
+ contract_inputs,
+ exclude_extension_id=exclude_extension_id,
+ patchable_paths=patchable_paths,
+ ),
+ contract_inputs,
+ )
diff --git a/tests/unit/test_extension_stage_lightweight_api.py b/tests/unit/test_extension_stage_lightweight_api.py
index 84cfaf349..37603b928 100644
--- a/tests/unit/test_extension_stage_lightweight_api.py
+++ b/tests/unit/test_extension_stage_lightweight_api.py
@@ -70,6 +70,96 @@ def _client(tmp_path: Path, service: _FakeService | None = None):
return TestClient(app), queued
+def test_upstream_candidates_endpoint_returns_job_relative_paths(
+ tmp_path: Path, monkeypatch
+) -> None:
+ from gfa_core.extension_stage_lightweight.api import upstream
+
+ workspace_root = tmp_path / "workspace"
+ job_root = workspace_root / "job_x" / "extension_runs"
+ output = job_root / "ext_up" / "runtime_r1" / "outputs" / "samples.tsv"
+ output.parent.mkdir(parents=True)
+ output.write_text("x\n", encoding="utf-8")
+ (job_root / "ext_up" / "runtime_r1" / "runtime_execution_result.json").write_text(
+ '{"success": true}', encoding="utf-8"
+ )
+
+ monkeypatch.setattr(
+ upstream,
+ "resolve_upstream_slots",
+ lambda workspace_root, job_id, contract_inputs, exclude_extension_id="", patchable_paths=None: {
+ "samples.tsv": upstream.UpstreamCandidate(
+ path=output,
+ relative="samples.tsv",
+ source_extension_id="ext_up",
+ run_id="runtime_r1",
+ is_dir=False,
+ run_time=0.0,
+ )
+ },
+ )
+
+ service = _FakeService()
+ service.help = lambda extension_id: {
+ "extension_id": extension_id,
+ "inputs": [
+ {"path": "samples.tsv", "replaceable": True},
+ {"path": "locked.tsv", "replaceable": False},
+ ],
+ }
+ client, _queued = _client(tmp_path, service)
+
+ response = client.get(
+ "/api/lightweight-extension-stages/count_fasta/upstream-candidates",
+ params={"job_id": "job_x"},
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+ assert data["extension_id"] == "count_fasta"
+ assert data["job_id"] == "job_x"
+ assert data["candidates"] == {
+ "samples.tsv": {
+ "path": "workspace/job_x/extension_runs/ext_up/runtime_r1/outputs/samples.tsv",
+ "source_extension_id": "ext_up",
+ "run_id": "runtime_r1",
+ "is_dir": False,
+ }
+ }
+
+
+def test_upstream_candidates_endpoint_empty_job_id_returns_empty(
+ tmp_path: Path,
+) -> None:
+ client, _queued = _client(tmp_path, _FakeService())
+
+ response = client.get(
+ "/api/lightweight-extension-stages/count_fasta/upstream-candidates"
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+ assert data["candidates"] == {}
+ assert data["job_id"] == ""
+
+
+def test_upstream_candidates_endpoint_unknown_extension_404(tmp_path: Path) -> None:
+ service = _FakeService()
+
+ def _help_missing(extension_id: str) -> dict:
+ raise ValueError("extension not found")
+
+ service.help = _help_missing
+ client, _queued = _client(tmp_path, service)
+
+ response = client.get(
+ "/api/lightweight-extension-stages/missing_ext/upstream-candidates",
+ params={"job_id": "job_x"},
+ )
+
+ assert response.status_code == 404
+
+
def _receipt(tmp_path: Path, *, token: str = "validate-1") -> None:
path = tmp_path / "store" / "operations" / f"{token}.json"
path.parent.mkdir(parents=True)
diff --git a/tests/unit/test_extension_stage_lightweight_frontend_contract.py b/tests/unit/test_extension_stage_lightweight_frontend_contract.py
index 3e686844d..1aa438469 100644
--- a/tests/unit/test_extension_stage_lightweight_frontend_contract.py
+++ b/tests/unit/test_extension_stage_lightweight_frontend_contract.py
@@ -300,3 +300,14 @@ def test_lightweight_nl_run_prefill_flag_wired() -> None:
assert "prefill" in source
assert "已识别为轻量拓展分析" not in source # 消息文案由后端生成，前端按 taskType 路由
+
+def test_lightweight_nl_run_wizard_loads_upstream_candidates() -> None:
+ source = _source()
+
+ assert "loadUpstreamCandidates" in source
+ assert "/upstream-candidates?job_id=" in source
+ assert "source_extension_id" in source
+ assert "上游样本" in source
+ assert "input.source" in source
+ assert "source: ''" in source
+
diff --git a/tests/unit/test_extension_stage_lightweight_nl_backend.py b/tests/unit/test_extension_stage_lightweight_nl_backend.py
index dd0a7442f..60e74af78 100644
--- a/tests/unit/test_extension_stage_lightweight_nl_backend.py
+++ b/tests/unit/test_extension_stage_lightweight_nl_backend.py
@@ -242,6 +242,28 @@ def test_job_file_hints_capped_at_max(monkeypatch, tmp_path: Path) -> None:
assert len(hints) == nl_intent.MAX_JOB_FILE_HINTS
+def test_job_file_hints_include_extension_runs_outputs(
+ monkeypatch, tmp_path: Path
+) -> None:
+ workspace_root = tmp_path / "workspace"
+ _write(
+ workspace_root,
+ "job-h1/extension_runs/ext_a/runtime_r1/outputs/out.tsv",
+ "a\tb\n",
+ )
+ monkeypatch.setattr(WEB_MAIN, "INPUT_ROOT", tmp_path / "input")
+ monkeypatch.setattr(WEB_MAIN, "WORKSPACE_ROOT", workspace_root)
+
+ hints = WEB_MAIN._lightweight_job_file_hints("job-h1")
+
+ assert "workspace/job-h1/extension_runs/ext_a/runtime_r1/outputs/out.tsv" in hints
+
+
+def test_nl_intent_prompt_mentions_upstream_extension_runs() -> None:
+ assert "extension_runs" in nl_intent._RESPONSE_SYSTEM_PROMPT
+ assert "runtime_*/outputs" in nl_intent._RESPONSE_SYSTEM_PROMPT
+
+
# ---------- 4. import-paths 端点 ----------
def _call_nl_import_paths(job_id: str, paths: list[str]):
diff --git a/tests/unit/test_extension_stage_lightweight_runtime_binding_resolver.py b/tests/unit/test_extension_stage_lightweight_runtime_binding_resolver.py
index a871e9bb8..b6ad4e17a 100644
--- a/tests/unit/test_extension_stage_lightweight_runtime_binding_resolver.py
+++ b/tests/unit/test_extension_stage_lightweight_runtime_binding_resolver.py
@@ -76,6 +76,44 @@ def _outputs(result: dict) -> dict[str, dict]:
return {item["fact_id"]: item for item in result["output_bindings"]}
+def test_resolves_directory_input(tmp_path: Path) -> None:
+ directory = tmp_path / "proteomes"
+ (directory / "01.fa").parent.mkdir(parents=True, exist_ok=True)
+ (directory / "01.fa").write_text(">a\nAAAA\n", encoding="utf-8")
+ (directory / "02.fa").write_text(">b\nBBBB\n", encoding="utf-8")
+ contract = _contract(
+ inputs=[_input("input:1", "proteomes/", "input directory")],
+ outputs=[("output:1", "result.txt")],
+ )
+
+ result = resolve_runtime_bindings(
+ contract, {"input:1": directory}, {}
+ )
+
+ assert result["status"] == "resolved"
+ assert result["input_bindings"] == [
+ {
+ "original_path": "proteomes/",
+ "runtime_path": str(directory),
+ "fact_ids": ["input:1"],
+ }
+ ]
+
+
+def test_resolves_missing_directory_reports_input_path_not_found(tmp_path: Path) -> None:
+ contract = _contract(
+ inputs=[_input("input:1", "proteomes/", "input directory")],
+ outputs=[("output:1", "result.txt")],
+ )
+
+ result = resolve_runtime_bindings(
+ contract, {"input:1": tmp_path / "missing_dir"}, {}
+ )
+
+ assert result["status"] != "resolved"
+ assert result["error"]["code"] == "INPUT_PATH_NOT_FOUND"
+
+
def test_resolves_single_input_and_output_prefix(tmp_path: Path) -> None:
replacement = _file(tmp_path, "Runtime.Sample.pep.fasta")
contract = _contract(
diff --git a/tests/unit/test_extension_stage_lightweight_runtime_package.py b/tests/unit/test_extension_stage_lightweight_runtime_package.py
index 4015f0ef0..572c2567c 100644
--- a/tests/unit/test_extension_stage_lightweight_runtime_package.py
+++ b/tests/unit/test_extension_stage_lightweight_runtime_package.py
@@ -1530,6 +1530,200 @@ def test_replay_isolates_symlink_input_before_source_overwrite(
assert result["execution_work_tree"]["Ft.input.txt"]["type"] == "regular"
+def _contract_and_resolved_dir(
+ raw_code: Path,
+ runtime_dir: Path,
+ *,
+ script: str,
+) -> tuple[dict, dict]:
+ """目录输入版的契约+解析构造：input 为目录，走 copy_tree 物化。"""
+ import shlex
+
+ from gfa_core.extension_stage_lightweight.runtime_binding import (
+ _sha256_tree,
+ )
+
+ script_path = _write(raw_code, "run.sh", script)
+ script_hash = _sha256_bytes(script_path.read_bytes())
+ commands = _prepare_script_commands(script, [])
+ targets = []
+ for index, command in enumerate(commands, start=1):
+ for token_index, token in enumerate(shlex.split(command)):
+ if token in {"proteomes", "proteomes/"}:
+ targets.append(
+ {
+ "binding_key": "path:proteomes",
+ "original_literal": token,
+ "kind": "input_path",
+ "script_path": "run.sh",
+ "script_hash": script_hash,
+ "expected_exact_match_count": 1,
+ "fact_ids": ["input:1"],
+ "anchors": [
+ {
+ "fact_id": "input:1",
+ "step_id": f"step_{index:03d}",
+ "command_digest": _sha256_bytes(command.encode()),
+ "token_index": token_index,
+ }
+ ],
+ }
+ )
+ contract = {
+ "schema_version": "lightweight-runtime-binding/v1",
+ "facts_sha256": "facts-locked",
+ "package_digest": "",
+ "original_command": "bash run.sh",
+ "source_scripts": [
+ {"path": "run.sh", "sha256": script_hash, "locked": True}
+ ],
+ "package_files": [
+ {
+ "path": "run.sh",
+ "source": "raw_code",
+ "kind": "file",
+ "size": script_path.stat().st_size,
+ "sha256": script_hash,
+ "locked": True,
+ "exposed_as_input": False,
+ },
+ ],
+ "inputs": [
+ {
+ "fact_id": "input:1",
+ "path": "proteomes",
+ "source": "baseline_input",
+ "replaceable": True,
+ "role": "input directory",
+ }
+ ],
+ "parameters": [],
+ "path_references": [],
+ "outputs": {
+ "files": [],
+ "stdout": {"present": True},
+ },
+ "patch_targets": targets,
+ "unpatchable_targets": [],
+ }
+ inventory = {
+ item["path"]: {
+ key: item[key]
+ for key in ("kind", "size", "sha256", "source")
+ }
+ for item in contract["package_files"]
+ }
+ contract["package_digest"] = _sha256_json(inventory)
+ _refresh_digest(contract, "contract_digest")
+ resolved = resolve_runtime_bindings(
+ contract,
+ {"input:1": runtime_dir},
+ {},
+ )
+ assert resolved["status"] == "resolved", resolved
+ return contract, resolved
+
+
+def test_render_materializes_directory_input_as_copy_tree(tmp_path: Path) -> None:
+ raw_code = tmp_path / "raw_code"
+ runtime_dir = tmp_path / "runtime_proteomes"
+ _write(runtime_dir, "01.fa", ">a\nAAAA\n")
+ _write(runtime_dir, "02.fa", ">b\nBBBB\n")
+ contract, resolved = _contract_and_resolved_dir(
+ raw_code,
+ runtime_dir,
+ script="ls proteomes\n",
+ )
+
+ rendered = render_runtime_package(
+ contract, raw_code, resolved, tmp_path / "run"
+ )
+
+ materialization = rendered["input_materialization"]
+ assert len(materialization) == 1
+ assert materialization[0]["method"] == "copy_tree"
+ assert materialization[0]["materialized_path"] == "runtime_proteomes"
+ assert len(materialization[0]["sha256"]) == 64
+ materialized = tmp_path / "run/work/runtime_proteomes"
+ assert materialized.is_dir()
+ assert not materialized.is_symlink()
+ assert (materialized / "01.fa").read_text(encoding="utf-8") == ">a\nAAAA\n"
+ patched = (tmp_path / "run/work/run.sh").read_text(encoding="utf-8")
+ assert "ls runtime_proteomes" in patched
+
+
+def test_replay_executes_directory_input_and_isolates_copy(
+ tmp_path: Path,
+) -> None:
+ import json
+
+ raw_code = tmp_path / "raw_code"
+ runtime_dir = tmp_path / "runtime_proteomes"
+ _write(runtime_dir, "01.fa", ">a\nAAAA\n")
+ _write(runtime_dir, "02.fa", ">b\nBBBB\n")
+ contract, resolved = _contract_and_resolved_dir(
+ raw_code,
+ runtime_dir,
+ script="ls proteomes\n",
+ )
+
+ rendered = render_runtime_package(
+ contract, raw_code, resolved, tmp_path / "run"
+ )
+ result = replay_runtime_binding(rendered)
+
+ assert result["contract_passed"] is True
+ assert result["stdout_contract_passed"] is True
+ # 执行期间工具写改 work 内副本不影响 source
+ assert (runtime_dir / "01.fa").read_text(encoding="utf-8") == ">a\nAAAA\n"
+ events = json.loads(
+ (tmp_path / "run/logs/execution_events.jsonl")
+ .read_text(encoding="utf-8")
+ .splitlines()[0]
+ )
+ assert events["execution_inputs"][0]["mode"] == "isolated_copy"
+ assert events["execution_inputs"][0]["materialized_path"] == "runtime_proteomes"
+
+
+def test_replay_rejects_directory_input_content_drift(tmp_path: Path) -> None:
+ raw_code = tmp_path / "raw_code"
+ runtime_dir = tmp_path / "runtime_proteomes"
+ _write(runtime_dir, "01.fa", ">a\nAAAA\n")
+ contract, resolved = _contract_and_resolved_dir(
+ raw_code,
+ runtime_dir,
+ script="ls proteomes\n",
+ )
+
+ rendered = render_runtime_package(
+ contract, raw_code, resolved, tmp_path / "run"
+ )
+ _write(runtime_dir, "03.fa", ">c\nCCCC\n") # 渲染后 source 被改动
+
+ with pytest.raises(RuntimeBindingError) as caught:
+ replay_runtime_binding(rendered)
+
+ assert caught.value.code == "INPUT_CONTENT_CHANGED"
+
+
+def test_sha256_tree_is_stable_and_skips_symlinks(tmp_path: Path) -> None:
+ from gfa_core.extension_stage_lightweight.runtime_binding import (
+ _sha256_tree,
+ )
+
+ source = tmp_path / "tree"
+ _write(source, "a.txt", "one\n")
+ _write(source, "sub/b.txt", "two\n")
+ first = _sha256_tree(source)
+ assert first == _sha256_tree(source)
+
+ (source / "link.txt").symlink_to(source / "a.txt")
+ assert _sha256_tree(source) == first # symlink 不参与摘要
+
+ _write(source, "sub/c.txt", "three\n")
+ assert _sha256_tree(source) != first # 新增文件改变摘要
+
+
def test_shell_word_replacement_preserves_python_string_literal_quotes() -> None:
"""zh01 回归：python -c 里带引号的输出路径被前缀替换后必须保留引号。
diff --git a/tests/unit/test_extension_stage_lightweight_service.py b/tests/unit/test_extension_stage_lightweight_service.py
index c351c8513..ed9ae7f66 100644
--- a/tests/unit/test_extension_stage_lightweight_service.py
+++ b/tests/unit/test_extension_stage_lightweight_service.py
@@ -168,6 +168,31 @@ def test_published_run_requires_job_context_and_projects_into_job_workspace(
)
+def test_job_path_accepts_directory_inside_job_and_rejects_outside(
+ tmp_path: Path,
+) -> None:
+ from gfa_core.extension_stage_lightweight.service import LightweightExtensionService
+
+ input_root = tmp_path / "input"
+ workspace_root = tmp_path / "workspace"
+ job_input = input_root / "job_1"
+ job_workspace = workspace_root / "job_1"
+ (job_input / "proteomes").mkdir(parents=True)
+ (job_workspace / ".keep").mkdir(parents=True)
+
+ inside = LightweightExtensionService._job_path(
+ job_input / "proteomes", (job_input, job_workspace)
+ )
+ assert inside.is_dir()
+
+ outside_dir = tmp_path / "outside"
+ outside_dir.mkdir()
+ with pytest.raises(ValueError, match="selected job"):
+ LightweightExtensionService._job_path(
+ outside_dir, (job_input, job_workspace)
+ )
+
+
def test_published_run_rejects_symlink_escape_from_selected_job(tmp_path: Path) -> None:
service = _published_copy_service(tmp_path)
input_root = tmp_path / "input"
diff --git a/tests/unit/test_extension_stage_lightweight_upstream.py b/tests/unit/test_extension_stage_lightweight_upstream.py
new file mode 100644
index 000000000..9e240893c
--- /dev/null
+++ b/tests/unit/test_extension_stage_lightweight_upstream.py
@@ -0,0 +1,329 @@
+"""轻量拓展分析"上游样本继承"发现模块单元测试。
+
+用 tmp_path 构造 workspace//extension_runs/ 假树（写入 runtime_execution_result.json
+的 success 字段与 outputs/ 产物），验证候选发现与槽位匹配的保守语义。
+"""
+from __future__ import annotations
+
+import json
+import os
+import time
+from pathlib import Path
+
+from gfa_core.extension_stage_lightweight import upstream
+
+
+def _write(root: Path, relative: str, content: str) -> Path:
+ path = root / relative
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(content, encoding="utf-8")
+ return path
+
+
+def _result(root: Path, success: bool) -> Path:
+ return _write(
+ root,
+ "runtime_execution_result.json",
+ json.dumps({"success": success, "contract_passed": success, "returncode": 0}),
+ )
+
+
+def _run_tree(
+ runs: Path,
+ ext_id: str,
+ run_id: str,
+ *,
+ outputs: list[str],
+ success: bool = True,
+) -> Path:
+ """建一个 run 目录并返回 run_dir；outputs 为相对 outputs/ 的文件路径。"""
+ run_dir = runs / ext_id / run_id
+ for relative in outputs:
+ _write(run_dir / "outputs", relative, "content:" + relative)
+ _result(run_dir, success)
+ return run_dir
+
+
+def _job_root(tmp_path: Path) -> Path:
+ workspace = tmp_path / "workspace"
+ _write(workspace, "job_1/.keep", "")
+ return workspace / "job_1" / "extension_runs"
+
+
+# ---------- 候选发现 ----------
+
+
+def test_collect_skips_failed_and_missing_result_runs(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_good", outputs=["a.tsv"])
+ _run_tree(runs, "ext_a", "runtime_failed", outputs=["b.tsv"], success=False)
+ missing = _run_tree(runs, "ext_a", "runtime_missing", outputs=["c.tsv"])
+ (missing / "runtime_execution_result.json").unlink()
+
+ pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
+
+ names = {candidate.path.name for candidate in pool}
+ assert names == {"a.tsv"}
+
+
+def test_collect_skips_excluded_extension(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_self", "runtime_1", outputs=["a.tsv"])
+ _run_tree(runs, "ext_other", "runtime_2", outputs=["b.tsv"])
+
+ pool = upstream.collect_job_upstream_outputs(
+ tmp_path / "workspace", "job_1", exclude_extension_id="ext_self"
+ )
+
+ sources = {candidate.source_extension_id for candidate in pool}
+ assert sources == {"ext_other"}
+
+
+def test_collect_ignores_non_runtime_run_dirs(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["a.tsv"])
+ # 标准拓展快照风格的 run_* 目录：glob("runtime_*") 天然不命中
+ snapshot = runs / "ext_b" / "run_20260826_120000"
+ _write(snapshot / "outputs", "legacy.tsv", "x")
+ _write(snapshot, "extension_run.json", json.dumps({"status": "success"}))
+
+ pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
+
+ names = {candidate.path.name for candidate in pool}
+ assert names == {"a.tsv"}
+
+
+def test_collect_empty_workspace_returns_empty(tmp_path: Path) -> None:
+ assert upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_none") == []
+ missing_ws = tmp_path / "no_workspace"
+ assert upstream.collect_job_upstream_outputs(missing_ws, "job_none") == []
+
+
+def test_collect_symlinks_are_skipped(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ run_dir = _run_tree(runs, "ext_a", "runtime_1", outputs=["real.tsv"])
+ (run_dir / "outputs" / "link.tsv").symlink_to(run_dir / "outputs" / "real.tsv")
+
+ pool = upstream.collect_job_upstream_outputs(tmp_path / "workspace", "job_1")
+
+ names = {candidate.path.name for candidate in pool}
+ assert names == {"real.tsv"}
+
+
+# ---------- 槽位匹配 ----------
+
+
+def test_exact_basename_match(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["samples.tsv"])
+ _run_tree(runs, "ext_b", "runtime_2", outputs=["other.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "samples.tsv", "replaceable": True}],
+ )
+
+ assert "samples.tsv" in found
+ (candidate,) = found["samples.tsv"]
+ assert candidate.path.name == "samples.tsv"
+ assert candidate.source_extension_id == "ext_a"
+
+
+def test_latest_run_wins(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["dup.tsv"])
+ older = _run_tree(runs, "ext_b", "runtime_old", outputs=["dup.tsv"])
+ stamp = time.time()
+ os.utime(newer, (stamp, stamp))
+ os.utime(older, (stamp - 100, stamp - 100))
+
+ selected = upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "dup.tsv", "replaceable": True}],
+ )
+
+ assert selected["dup.tsv"].run_id == "runtime_new"
+
+
+def test_shortest_path_wins_within_same_run(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["deep/nested/dup.tsv", "dup.tsv"])
+
+ selected = upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "dup.tsv", "replaceable": True}],
+ )
+
+ assert selected["dup.tsv"].relative == "dup.tsv"
+
+
+def test_ambiguous_role_match_yields_no_candidate(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["apple_output.tsv"])
+ _run_tree(runs, "ext_b", "runtime_2", outputs=["banana_output.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "wanted.tsv", "role": "output", "replaceable": True}],
+ )
+
+ assert "wanted.tsv" not in found
+
+
+def test_unique_role_match_yields_candidate(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["q1.tsv"])
+ _run_tree(runs, "ext_b", "runtime_2", outputs=["homology_unique_name.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "wanted.tsv", "role": "homology", "replaceable": True}],
+ )
+
+ assert "wanted.tsv" in found
+ assert found["wanted.tsv"][0].path.name == "homology_unique_name.tsv"
+
+
+def test_multi_token_role_requires_all_tokens(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["homology_hits.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ # "homology" 唯一命中，但 "absentword" 零命中 → 不全命中，不采纳
+ [{"path": "wanted.tsv", "role": "homology absentword", "replaceable": True}],
+ )
+
+ assert "wanted.tsv" not in found
+
+
+def test_role_multi_hit_takes_latest_run(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["fresh.hits.tsv"])
+ older = _run_tree(runs, "ext_b", "runtime_old", outputs=["stale.hits.tsv"])
+ stamp = time.time()
+ os.utime(newer, (stamp, stamp))
+ os.utime(older, (stamp - 100, stamp - 100))
+
+ selected = upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "wanted.tsv", "role": "hmm hits", "replaceable": True}],
+ )
+
+ assert "wanted.tsv" in selected
+ assert selected["wanted.tsv"].path.name == "fresh.hits.tsv"
+ assert selected["wanted.tsv"].run_id == "runtime_new"
+
+
+def test_unpatchable_slot_skips_role_match(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["fresh.hits.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "hmm_hits.tsv", "role": "hmm hits", "replaceable": True}],
+ patchable_paths={"01.col.primary.pep.fa"}, # 槽位不在可补丁集
+ )
+
+ assert "hmm_hits.tsv" not in found
+
+
+def test_exact_name_match_ignores_patchable_filter(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["samples.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "samples.tsv", "role": "manifest", "replaceable": True}],
+ patchable_paths=set(), # 精确同名不受可补丁过滤影响
+ )
+
+ assert "samples.tsv" in found
+
+
+def test_directory_slot_matches_directory_not_file(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["proteomes/01.fa", "proteomes/02.fa"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "proteomes/", "replaceable": True, "role": "input directory"}],
+ )
+
+ assert "proteomes/" in found
+ (candidate,) = found["proteomes/"]
+ assert candidate.is_dir is True
+ assert candidate.path.name == "proteomes"
+
+
+def test_file_slot_ignores_same_named_directory(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["03_homology/x.tsv"])
+
+ found = upstream.find_upstream_candidates(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "03_homology", "replaceable": True, "role": "homology table"}],
+ )
+
+ assert "03_homology" not in found
+
+
+def test_replaceable_filter_and_candidate_exclusivity(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ _run_tree(runs, "ext_a", "runtime_1", outputs=["shared.tsv"])
+
+ selected = upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [
+ {"path": "shared.tsv", "replaceable": True},
+ {"path": "shared.tsv", "replaceable": True},
+ {"path": "locked.tsv", "replaceable": False},
+ ],
+ )
+
+ assert set(selected) == {"shared.tsv"} # 同 path 第二个槽被占用后不再填；locked 不参与
+
+
+def test_duplicate_slot_path_keeps_latest_run(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path)
+ newer = _run_tree(runs, "ext_a", "runtime_new", outputs=["hit.tsv"])
+ older = _run_tree(runs, "ext_a", "runtime_old", outputs=["hit.tsv"])
+ stamp = time.time()
+ os.utime(newer, (stamp, stamp))
+ os.utime(older, (stamp - 100, stamp - 100))
+
+ selected = upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [
+ {"path": "hit.tsv", "replaceable": True},
+ {"path": "hit.tsv", "replaceable": True}, # 契约中同名不同 fact_id 的重复槽
+ ],
+ )
+
+ assert selected["hit.tsv"].run_id == "runtime_new"
+
+
+def test_no_upstream_returns_empty(tmp_path: Path) -> None:
+ runs = _job_root(tmp_path) # extension_runs 目录不存在
+ assert (
+ upstream.resolve_upstream_slots(
+ tmp_path / "workspace",
+ "job_1",
+ [{"path": "samples.tsv", "replaceable": True}],
+ )
+ == {}
+ )
+ assert not runs.exists()