Upload 15 files

src/model_scan.py

@@ -69,6 +69,197 @@ IMAGE_OUTPUT_HINT_RE = re.compile(
     r"(\.images\s*\[|generated image|text-to-image|image = pipe\()",
     re.IGNORECASE,
 )
+INPUT_MODALITY_ORDER = ("text", "image", "audio", "video", "file")
+INPUT_MODALITY_HINTS = {
+    "text": (
+        "text prompt",
+        "input prompt",
+        "prompt:",
+        "prompt=",
+        "gr.textbox",
+        "textbox",
+    ),
+    "image": (
+        "image_path",
+        "input image",
+        "image input",
+        "source image",
+        "reference image",
+        "avatar image",
+        "person image",
+        "gr.image",
+        "pil.image",
+        "upload image",
+    ),
+    "audio": (
+        "audio_path",
+        "input audio",
+        "audio input",
+        "source audio",
+        "speech audio",
+        "voice audio",
+        "single audio",
+        "multi audio",
+        "gr.audio",
+        ".wav",
+    ),
+    "video": (
+        "video_path",
+        "input video",
+        "video input",
+        "source video",
+        "reference video",
+        "gr.video",
+        ".mp4",
+    ),
+    "file": (
+        "file_path",
+        "input file",
+        "file input",
+        "upload file",
+        "gr.file",
+    ),
+}
+
+
+def _ordered_modalities(values: set[str]) -> list[str]:
+    return [name for name in INPUT_MODALITY_ORDER if name in values]
+
+
+def infer_expected_inputs(
+    *,
+    pipeline_tag: str | None = None,
+    tags: list[str] | None = None,
+    readme: str | None = None,
+) -> list[str]:
+    modalities: set[str] = set()
+    task = (pipeline_tag or "").strip().lower()
+    tag_set = {str(t).lower() for t in (tags or [])}
+    task_like = {task, *tag_set}
+
+    for item in task_like:
+        if item.startswith("text-to-") or item in {
+            "text-generation",
+            "text2text-generation",
+            "summarization",
+            "translation",
+            "question-answering",
+            "fill-mask",
+            "sentence-similarity",
+            "token-classification",
+            "zero-shot-classification",
+        }:
+            modalities.add("text")
+        if item.startswith("image-to-") or item in {"image-classification", "object-detection", "image-segmentation"}:
+            modalities.add("image")
+        if item.startswith("audio-to-") or item in {"automatic-speech-recognition", "audio-classification"}:
+            modalities.add("audio")
+        if item.startswith("video-to-"):
+            modalities.add("video")
+
+    text = (readme or "").lower()
+    for modality, hints in INPUT_MODALITY_HINTS.items():
+        if any(hint in text for hint in hints):
+            modalities.add(modality)
+    if re.search(r"\bprompt\b", text) and any(word in text for word in ("pipe(", "generate", "demo", "gradio")):
+        modalities.add("text")
+
+    return _ordered_modalities(modalities)
+
+
+def infer_output_guess(
+    *,
+    pipeline_tag: str | None = None,
+    library_name: str | None = None,
+    tags: list[str] | None = None,
+    pipeline_class: str | None = None,
+    readme: str | None = None,
+) -> dict[str, Any]:
+    expected = infer_expected_output_type(
+        pipeline_tag=pipeline_tag,
+        library_name=library_name,
+        tags=tags,
+        pipeline_class=pipeline_class,
+        readme=readme,
+    )
+    task = (pipeline_tag or "").strip().lower()
+    tag_set = {str(t).lower() for t in (tags or [])}
+    cls = (pipeline_class or "").strip().lower()
+    text = (readme or "").lower()
+    library = (library_name or "").strip().lower()
+    combined = " ".join([library, cls, " ".join(sorted(tag_set)), text])
+    evidence: list[str] = []
+    confidence = "low"
+    source = "none"
+
+    if task in TASK_OUTPUT_TYPE_MAP:
+        confidence = "high"
+        source = "pipeline_tag"
+        evidence.append(f"pipeline_tag={task} maps to {TASK_OUTPUT_TYPE_MAP[task]}.")
+    elif {"text-to-video", "image-to-video"} & tag_set:
+        confidence = "high"
+        source = "tags"
+        evidence.append("Hub tags include text-to-video or image-to-video.")
+    elif {"text-to-image", "image-to-image", "diffusers"} & tag_set and expected == "image":
+        confidence = "medium"
+        source = "tags"
+        evidence.append("Hub tags indicate a Diffusers image workflow.")
+    elif {"text-to-audio", "text-to-speech", "audio"} & tag_set and expected == "audio":
+        confidence = "medium"
+        source = "tags"
+        evidence.append("Hub tags indicate an audio generation workflow.")
+    elif any(hint in cls for hint in IMAGE_PIPELINE_HINTS) and expected == "image":
+        confidence = "medium"
+        source = "pipeline_class"
+        evidence.append(f"Pipeline class {pipeline_class} looks image-oriented.")
+    elif "video" in cls and expected == "video":
+        confidence = "high"
+        source = "pipeline_class"
+        evidence.append(f"Pipeline class {pipeline_class} looks video-oriented.")
+    elif "audio" in cls or "speech" in cls:
+        confidence = "medium"
+        source = "pipeline_class"
+        evidence.append(f"Pipeline class {pipeline_class} looks audio-oriented.")
+
+    if VIDEO_OUTPUT_HINT_RE.search(combined) or re.search(r"\.frames\s*\[", text):
+        if expected == "video":
+            confidence = "high"
+            source = "model_card"
+        evidence.append("Model card or library name mentions video/avatar generation markers.")
+    elif IMAGE_OUTPUT_HINT_RE.search(text) and expected == "image":
+        if confidence == "low":
+            confidence = "medium"
+            source = "model_card"
+        evidence.append("Model card includes image output usage such as pipe(...).images.")
+    elif re.search(r"\.audios?\s*\[|generated audio|text-to-speech|soundfile|\.wav", text) and expected == "audio":
+        if confidence == "low":
+            confidence = "medium"
+            source = "model_card"
+        evidence.append("Model card includes audio output markers.")
+    elif re.search(r"generated_text|tokenizer\.decode|text-generation|response\s*=", text) and expected == "text":
+        if confidence == "low":
+            confidence = "medium"
+            source = "model_card"
+        evidence.append("Model card includes text output markers.")
+
+    if expected == "any":
+        confidence = "low"
+        if not evidence:
+            evidence.append("Custom or ambiguous library without a strong generated-output signal; keeping output flexible.")
+    elif not expected:
+        evidence.append("No strong generated-output signal found in pre-scan metadata.")
+    elif not evidence:
+        evidence.append(f"Pre-scan inferred {expected} from weak metadata signals.")
+
+    return {
+        "schema_version": "pre_scan_output_guess.v1",
+        "expected_output_type": expected or "",
+        "confidence": confidence,
+        "source": source,
+        "authority": "pre_scan_guess",
+        "final_authority": "runtime_contract_or_gradio_schema",
+        "evidence": evidence[:6],
+    }
 
 
 def infer_expected_output_type(
@@ -429,13 +620,19 @@ def analyze_model_metadata(
     has_diffusers_example = bool(card_signals["has_diffusers_example"])
     pipeline_class = str(card_signals["pipeline_class"] or "")
     runtime_hints = list(card_signals["runtime_hints"] or [])
-    expected_output_type = infer_expected_output_type(
+    output_guess = infer_output_guess(
         pipeline_tag=pipeline_tag,
         library_name=library_name,
         tags=tags,
         pipeline_class=pipeline_class,
         readme=readme,
     )
+    expected_output_type = output_guess["expected_output_type"] or None
+    expected_inputs = infer_expected_inputs(
+        pipeline_tag=pipeline_tag,
+        tags=tags,
+        readme=readme,
+    )
 
     score = 50
     good: list[str] = []
@@ -457,8 +654,10 @@ def analyze_model_metadata(
     else:
         score -= 12
         risk.append("No pipeline tag detected; task type may be ambiguous for an autonomous builder.")
+    if expected_inputs:
+        good.append("Inputs detected by pre-scan: " + ", ".join(expected_inputs) + ".")
     if expected_output_type:
-        good.append(f"Expected output type inferred: {expected_output_type}.")
+        good.append(f"Pre-scan output guess: {expected_output_type} (confidence {output_guess['confidence']}).")
 
     if library_name:
         score += 10
@@ -599,6 +798,11 @@ def analyze_model_metadata(
         "risk_signals": risk[:10],
         "recommendations": recommendations[:6],
         "expected_output_type": expected_output_type or "",
+        "expected_inputs": expected_inputs,
+        "output_confidence": output_guess["confidence"],
+        "output_authority": output_guess["authority"],
+        "output_evidence": output_guess["evidence"],
+        "pre_scan_output": output_guess,
         "build_risk": build_risk,
         "kernel_strategy": kernel_strategy,
         "metadata": {
@@ -616,6 +820,11 @@ def analyze_model_metadata(
             "pipeline_class": pipeline_class,
             "runtime_hints": runtime_hints,
             "expected_output_type": expected_output_type or "",
+            "expected_inputs": expected_inputs,
+            "output_confidence": output_guess["confidence"],
+            "output_authority": output_guess["authority"],
+            "output_evidence": output_guess["evidence"],
+            "pre_scan_output": output_guess,
             "diffusers_standard": bool(is_diffusers and has_model_index and has_safetensors),
             "build_risk_level": build_risk["level"],
             "recommended_session_minutes": build_risk["recommended_session_minutes"],

src/version.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-ASF_APP_VERSION = "v198.26.60"
-ASF_RELEASE_NAME = "Agentic Space Factory v198.26.60"
+ASF_APP_VERSION = "v198.26.61"
+ASF_RELEASE_NAME = "Agentic Space Factory v198.26.61"
 
 
 def resolve_app_version(value: str | None = None) -> str:

src/worker_payload.py

@@ -39,8 +39,8 @@ DEFAULT_PREFERRED_SPACE_HARDWARE = "zero-a10g"
 DEFAULT_FALLBACK_SPACE_HARDWARE = "a10g-large"
 DEFAULT_MODEL_ID = "sshleifer/tiny-gpt2"
 MAX_PI_REPAIR_ATTEMPTS = 3
-APP_VERSION = "v198.26.60"
-app_version = "v198.26.60"
+APP_VERSION = "v198.26.61"
+app_version = "v198.26.61"
 
 # Internal agent/recovery files may be needed inside the transient Pi
 # workspace, but they should not be published to the generated Space or shown
@@ -18892,7 +18892,7 @@ def smoke_generate(target_space_id: str, token: str, run_dir: Path, events_path:
     # write_json(run_dir / "tests" / "payload_source.json", payload_source_record)
     # write_json(run_dir / "tests" / "replay_source.json", replay_source)
     # write_json(run_dir / "tests" / "validation_preflight.json", validation_preflight)
-    app_version = "v198.26.60"
+    app_version = "v198.26.61"
     engine_version = "unified_gradio_validation_harness_v198_25_3"
     parent_build_run_id = os.environ.get("PARENT_BUILD_RUN_ID", "").strip()
     validation_mode = os.environ.get("VALIDATION_MODE") or os.environ.get("SPACE_TEST_POLICY_MODE") or "linked"