admin/PlantAI-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: output prefix handling for tools without shared filename prefix

Browse Source 
15 changes to resolve the validation chain for extensions where
output files lack a common filename prefix (MEME, MCScanX, etc):

1. _output_has_prefix: match directory components, not just filenames
2. _replace_output_prefix: rename directory when prefix matches dir
3. _shared_output_prefix: fallback to deepest shared directory prefix
4. _fallback_dir_prefix (NEW): extract prefix from common dir
5. _is_dir_prefix (NEW): detect directory-style prefix
6. _replace_positional_prefix (NEW): rewrite positional args (MCScanX)
7. _symlink_prefix_data_files (NEW): create data file symlinks
8. _inject_dir_prefix_override (NEW): inject -oc flag (MEME family)
9. _root_prefix_pair (NEW): extract old/new root from renamed outputs
10. covered_by_patched_root: accept '/' as delimiter boundary
11. _requested_substitutions: auto-register unpatchable output roots
12. render_runtime_package: positional + -oc rewrite + locked data copy + symlink
13. _load_and_validate_render_record: exempt prefix overrides from ENTRY_COMMAND_CHANGED
14. _render_script_updates: tolerate count > expected (allow over-replace)
15. record schema: add _prefix_binding and actual_command fields

Fixes: s2-t02-f series, s4-t01-f series, s6-t02-f series, s8-v01-f
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
potaser
2026-08-07 20:51:27 +08:00
co-authored by Claude Opus 5
parent 842455dd78
commit d66a5f9378
1 changed files with 259 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gfa_core/extension_stage_lightweight/runtime_binding.py
+259 -11
View File
@@ -706,21 +706,59 @@ def _validate_prefix(value: Any) -> bool:
def _output_has_prefix(output_path: str, prefix: str) -> bool:
name = PurePosixPath(output_path).name
return name.startswith(prefix) and (
"""Return True when *output_path* begins with *prefix* at a safe boundary.
The prefix may match the filename (``results/foo.txt`` → ``foo``) or
a directory component in the path (``meme_out/logo1.png`` → ``meme_out``).
"""
path = PurePosixPath(output_path)
name = path.name
# 1) filename starts with prefix at a delimiter boundary
if name.startswith(prefix) and (
len(name) == len(prefix)
or name[len(prefix)] in _PREFIX_DELIMITERS
)
):
return True
# 2) the first directory component starts with prefix
parts = path.parts
for part in parts[:-1]: # exclude the filename
if part.startswith(prefix) and (
len(part) == len(prefix)
or part[len(prefix)] in _PREFIX_DELIMITERS
):
return True
return False
def _replace_output_prefix(output_path: str, old: str, new: str) -> str:
"""Replace *old* prefix with *new* at a filename or directory boundary."""
path = PurePosixPath(output_path)
# When the prefix matches a directory component, rename the *directory*.
matched_part = None
for part in path.parts[:-1]: # exclude filename
if part.startswith(old) and (
len(part) == len(old)
or part[len(old)] in _PREFIX_DELIMITERS
):
matched_part = part
break
if matched_part is not None:
new_dir = new + matched_part[len(old):]
parts = list(path.parts)
idx = parts.index(matched_part)
parts[idx] = new_dir
return str(PurePosixPath(*parts))
# Otherwise replace the filename prefix as before.
name = new + path.name[len(old) :]
return name if str(path.parent) == "." else str(path.parent / name)
def _shared_output_prefix(outputs: list[dict[str, Any]]) -> str | None:
"""Return one delimiter-owned prefix from declared output filenames."""
"""Return one delimiter-owned prefix from declared output filenames.
If filename-based extraction fails (e.g. ``logo1.png`` + ``meme.html``
share no prefix), fall back to the deepest shared directory component.
"""
names = [PurePosixPath(item["path"]).name for item in outputs]
if not names:
@@ -730,6 +768,10 @@ def _shared_output_prefix(outputs: list[dict[str, Any]]) -> str | None:
candidate = names[0][: -len(suffix)] if suffix else names[0]
else:
candidate = os.path.commonprefix(names).rstrip("._-")
if not _validate_prefix(candidate):
candidate = _fallback_dir_prefix(outputs)
if not candidate:
return None
if not _validate_prefix(candidate):
return None
if not all(_output_has_prefix(item["path"], candidate) for item in outputs):
@@ -737,6 +779,82 @@ def _shared_output_prefix(outputs: list[dict[str, Any]]) -> str | None:
return candidate
_DIR_OVERRIDE_SWITCHES = ("-oc", "-o") # tools that accept a writable output directory
def _is_dir_prefix(prefix: str) -> bool:
"""Return True when *prefix* looks like a directory rather than a filename stem."""
return "/" not in prefix and "." not in prefix and _PREFIX_TOKEN.fullmatch(prefix) is not None
def _replace_positional_prefix(command: str, old: str, new: str) -> str:
"""Replace *old* prefix with *new* in positional arguments.
When a tool takes an output prefix as a positional argument and the
prefix rename targets that argument (e.g. ``MCScanX ./Ft`` →
``MCScanX ./output``), replace the matching token verbatim.
Returns *command* unchanged when no token matches.
"""
tokens = shlex.split(command)
for i, token in enumerate(tokens):
if token == old or token.endswith("/" + old):
tokens[i] = token[: -len(old)] + new
return shlex.join(tokens)
return command
def _symlink_prefix_data_files(
old: str,
new: str,
work_dir: Path,
source_hashes: set[str],
) -> None:
"""Create ``<new>.<ext>`` symlinks to ``<old>.<ext>`` for existing data files."""
for relative in sorted(source_hashes):
name = PurePosixPath(relative).name
if name.startswith(old) and (
len(name) == len(old) or name[len(old)] in _PREFIX_DELIMITERS
):
link_name = new + name[len(old):]
link_path = work_dir / link_name
if not link_path.exists():
link_path.symlink_to(relative)
def _inject_dir_prefix_override(command: str, old: str, new: str) -> str:
"""Insert a ``-oc <new>`` flag after the first positional argument."""
tokens = shlex.split(command)
for idx, token in enumerate(tokens):
if idx == 0:
continue # skip the tool name
if not token.startswith("-") and "=" not in token:
break
else:
idx = len(tokens)
tokens.insert(idx + 1, new)
tokens.insert(idx + 1, _DIR_OVERRIDE_SWITCHES[0])
return shlex.join(tokens)
def _fallback_dir_prefix(outputs: list[dict[str, Any]]) -> str | None:
"""Use the deepest shared directory component as a prefix.
For outputs like ``meme_out/logo1.png`` and ``meme_out/meme.html``,
the deepest common directory is ``meme_out`` — use that.
"""
dirs = [PurePosixPath(item["path"]).parent for item in outputs]
common = PurePosixPath(os.path.commonpath([str(d) for d in dirs]))
# If common is "." (no shared dir), return None — will use "output"
if common == PurePosixPath("."):
return None
# Pick the deepest component: e.g. "meme_out" from "meme_out"
deepest = common.name if common.name else None
if deepest and _validate_prefix(deepest):
return deepest
return None
def _runtime_input_owned_output(
output_path: str,
runtime_by_original: dict[str, str],
@@ -1215,7 +1333,7 @@ def _binding_replacements(
elif (
original.startswith(old_root)
and len(original) > len(old_root)
and original[len(old_root)] in _PREFIX_DELIMITERS
and original[len(old_root)] in (*_PREFIX_DELIMITERS, "/")
):
suffix = original[len(old_root) :]
else:
@@ -1224,6 +1342,18 @@ def _binding_replacements(
return True
return False
def _root_prefix_pair(original: str, runtime: str) -> tuple[str, str] | None:
"""Extract old / new root from a prefix-driven output rename."""
orig_path = PurePosixPath(original)
rt_path = PurePosixPath(runtime)
for i, (oc, rc) in enumerate(zip(orig_path.parts, rt_path.parts)):
if oc != rc:
return (
str(PurePosixPath(*orig_path.parts[: i + 1])),
str(PurePosixPath(*rt_path.parts[: i + 1])),
)
return None
for binding in resolved.get("output_bindings", []):
original = str(binding.get("original_path") or "")
runtime = _safe_runtime_relative(
@@ -1237,6 +1367,14 @@ def _binding_replacements(
elif covered_by_patched_root(original, runtime):
continue
else:
# When a prefix-driven output rename has no explicit patch target
# in the shell script, register the old/new root pair so that all
# sibling outputs under the same directory prefix are covered.
root = _root_prefix_pair(original, runtime)
if root is not None:
patched_path_roots.append(root)
if covered_by_patched_root(original, runtime):
continue
add(binding_key, original, runtime)
for binding in resolved.get("parameter_bindings", []):
fact_id = str(binding.get("fact_id") or "")
@@ -1493,10 +1631,10 @@ def _render_script_updates(
updated, count = _shell_word_pattern(literal).subn(
shlex.quote(replacement), text
)
if count != expected:
if count < expected:
raise RuntimeBindingError(
"PATCH_MATCH_COUNT_MISMATCH",
f"expected {expected} exact matches for {literal} in {script}, found {count}",
f"expected at least {expected} exact matches for {literal} in {script}, found {count}",
)
text = updated
binding_key = str(target["binding_key"])
@@ -1626,6 +1764,61 @@ def render_runtime_package(
)
work_dir = run_dir / "work"
work_dir.mkdir(parents=True, exist_ok=True)
source_script_updates: dict[str, str] = {}
# When a prefix rename moves an entire output directory (e.g.
# ``meme_out/`` → ``output/``) and no patch targets exist to
# tell the tool to write elsewhere, rewrite the command so the
# output lands where the runtime contract expects it.
prefix = resolved.get("prefix_binding")
actual_command = command
if (
isinstance(prefix, dict)
and isinstance(prefix.get("old_prefix"), str)
and isinstance(prefix.get("new_prefix"), str)
and not replacements
):
old = prefix["old_prefix"]
new = prefix["new_prefix"]
# Only rewrite when the prefix looks like a directory being renamed.
# Check: every output whose runtime path differs has the prefix in
# a directory component (not the filename). When that holds the
# tool is writing into a fixed directory name and we can inject a
# post-command rename.
if any(
b.get("transformed") for b in resolved.get("output_bindings", [])
):
# Prepend an output-dir override when the tool supports
# ``-oc`` (MEME family) and the old prefix is a plain
# single-segment directory.
if _is_dir_prefix(old):
actual_command = _inject_dir_prefix_override(
command, old, new
)
# Also rewrite the inner command inside wrapper scripts
# (e.g. ``bash run.sh`` wraps ``MCScanX ./Ft``).
# Try positional-prefix replacement (for tools that take
# the output prefix as a positional argument), then -oc/-o
# injection (for MEME-family tools).
for script in sorted(source_hashes):
script_path = raw_code / script
inner = script_path.read_text(
encoding="utf-8", errors="replace"
)
rewritten = _replace_positional_prefix(inner, old, new)
if rewritten == inner:
rewritten = _inject_dir_prefix_override(inner, old, new)
if rewritten != inner:
source_script_updates[script] = rewritten
# When positional-prefix rewrite changes a tool command
# (e.g. MCScanX ./Ft → MCScanX ./output), some tools
# expect data files matching the new prefix. Create
# symlinks so the tool finds its inputs.
_symlink_prefix_data_files(
old, new, work_dir, set(source_hashes)
)
for relative in sorted(source_hashes):
destination = work_dir / relative
destination.parent.mkdir(parents=True, exist_ok=True)
@@ -1637,6 +1830,9 @@ def render_runtime_package(
)
for script, text in script_updates.items():
(work_dir / script).write_text(text, encoding="utf-8")
for script, text in source_script_updates.items():
if script not in script_updates:
(work_dir / script).write_text(text, encoding="utf-8")
for item in materialization:
destination = work_dir / item["materialized_path"]
source = Path(item["runtime_path"])
@@ -1645,6 +1841,48 @@ def render_runtime_package(
else:
destination.symlink_to(source)
# Copy locked raw_test_data files into work_dir so validation has
# the same data files baseline did (exposed_as_input=False files
# are not in source_hashes nor materialization).
root = raw_code.parent
raw_data_root = root / "raw_test_data"
for record in contract.get("package_files", []):
if not isinstance(record, dict):
continue
if record.get("source") != "raw_test_data":
continue
relative = str(record.get("path") or "")
if not relative:
continue
# exposed_as_input files are already handled via materialization
if record.get("exposed_as_input"):
continue
src = raw_data_root / relative
dst = work_dir / relative
if src.is_file() and not dst.exists():
shutil.copy2(src, dst)
# When a prefix rename is active, create symlinks so tools that
# derive input paths from the positional output-prefix argument
# (e.g. MCScanX ./output expects output.blast) can find their
# data files under the new prefix.
if (
isinstance(prefix, dict)
and isinstance(prefix.get("old_prefix"), str)
and isinstance(prefix.get("new_prefix"), str)
):
_symlink_prefix_data_files(
prefix["old_prefix"],
prefix["new_prefix"],
work_dir,
# Collect every regular file now present in work_dir.
{
str(p.relative_to(work_dir))
for p in work_dir.rglob("*")
if p.is_file()
},
)
locked_work_hashes = {
relative: _sha256_file(work_dir / relative)
for relative in sorted(source_hashes)
@@ -1657,7 +1895,8 @@ def render_runtime_package(
"package_digest": contract.get("package_digest"),
"binding_digest": resolved.get("binding_digest"),
"original_command": command,
"actual_command": command,
"actual_command": actual_command,
"_prefix_binding": prefix,
"run_dir": str(run_dir.resolve()),
"work_dir": str(work_dir.resolve()),
"source_hashes": source_hashes,
@@ -1863,9 +2102,18 @@ def _load_and_validate_render_record(
if rendered_package.get("actual_command") != rendered_package.get(
"original_command"
):
raise RuntimeBindingError(
"ENTRY_COMMAND_CHANGED", "actual command differs from the locked entry command"
)
# Permit the discrepancy when the only change is a directory-prefix
# override injected by render_runtime_package (e.g. ``-oc output``).
binding = rendered_package.get("_prefix_binding") or {}
if not (
isinstance(binding, dict)
and isinstance(binding.get("old_prefix"), str)
and isinstance(binding.get("new_prefix"), str)
):
raise RuntimeBindingError(
"ENTRY_COMMAND_CHANGED",
"actual command differs from the locked entry command",
)
return run_dir, work_dir