[api][runtime] Extend agent skill loading repository.

[wenjin272]

Linked issue: #592, #593, #594

Purpose of change

Extend the way Flink-Agents loads external agent skills:

• load agent skills from remote urls (only http/https point to a zip)
• load agent skills from python package
• load agent skills from java classpath

Tests

ut

API

Yes, add from_xxx to Skills.

Documentation

• doc-needed
• doc-not-needed
• doc-included

[wzhero1]

Thanks for the PR. Left some comments as below.

[wzhero1]

Inheritance is being used as composition.

A URL repo is not logically a FileSystem repo — it fetches over the network, then delegates. Side effects: the constructor performs network I/O before super(...), the original url is lost for diagnostics, there's no close() lifecycle, and the fetch step can't be mocked alone.

Suggest a SkillSource { Path materialize() } interface composed into one MaterializingSkillRepository. Same applies to ClasspathSkillRepository and PackageSkillRepository.

[wzhero1]

Adding a new source today forces 5 parallel changes (Skills field + factory + XxxSkillRepository + SkillManager for-loop + AgentPlan merge branch).

Suggest a discriminated union — single sources: List<SkillSourceSpec(scheme, location)> field + a scheme → handler registry. New source = one registration, no other file touched. Bonus: cross-language ser/de becomes uniform.

Wire-format-affecting; much cheaper to fix before 0.3.0 ships.

[wzhero1]

Cross-language deserialization silently drops fields. Python writes packages → Java drops it → 0 skills loaded, no warning. Mirror for Java classpathResources → Python.

Short term: emit a WARN on unrecognized source fields during deserialization. Long term: unify under one cross-language scheme (see the sources discriminator suggested on Skills.java).

[wzhero1]

⚠️ Flink fat-jar / nested-jar deployment is not covered.

Flink user JARs typically use Maven Shade or jar-in-jar; getResource("skills") returns protocols like jar:nested: that throw right here. findInUrlClassLoader also assumes URLClassLoader, not Flink's ChildFirstClassLoader.

Issue #593 explicitly targets this scenario. Before merge: (1) an end-to-end test on a Flink mini-cluster, (2) use ClassLoader.getResources() + getResourceAsStream instead of dispatching on URL protocol.

[wenjin272]

What is the significance of highlighting URLClassLoader and ChildFirstClassLoader here? I understand that ChildFirstClassLoader is also a subclass of URLClassLoader.

Since we set the contextClassLoader to Flink's UserCodeClassloader before executing a JavaAction, it is used in most scenarios. However, there was an issue with the initialization of the skill-managed path in Python Actions. I have modified the code to explicitly pass the UserCodeClassloader.

Typically, I understand that the Flink user JAR does not use the jar-in-jar approach.

getResourceAsStream requires a file path as input, but the resource here is typically a directory name, so getResourceAsStream cannot be used.

[wzhero1]

Constructor is synchronous, blocking, and not recoverable. loadAll() fetches sources serially; one slow URL stalls startup, one failure kills the whole SkillManager, and repos is final so there's no hot reload.

Follow-up: parallel materialization (CompletableFuture), fail-soft mode with a failure list, lazy load on first getSkill(), optional reloadSource() API.

[wenjin272]

This behavior will be optimized in subsequent iterations.

[wzhero1]

Duplicate skill names overwrite silently with no log. Multi-source name collisions go through skills.put() and the later registration wins — no WARN, no error. Users debugging "why doesn't my SKILL.md take effect" get no diagnostic.

Minimum: WARN with both origins. Better: document the paths → urls → classpath precedence. (Naming the source concretely needs the SkillOrigin change on AgentSkill.)

[wzhero1]

findInUrlClassLoader silently uses only the first matching jar. Multiple plugin jars with the same prefix → every jar after the first is dropped, no log. Minimum: log the choice. Ideally merge across all matches, or fail loudly when more than one matches.

Also assumes URLClassLoader, not Flink's ChildFirstClassLoader — see the nested-jar thread on materialize.

[wenjin272]

I think merge across all matches makes sense.

[wzhero1]

Source-incompatible constructor change. Wire compat is fine; but new Skills(List.of("/data/skills")) no longer compiles. Either keep a @Deprecated 1-arg constructor that forwards, or call it out in the 0.3.0 release notes.

If Skills later migrates to a single sources field (see the discriminator thread), this signature changes again — better to break once.

[wzhero1]

Merge-coverage gap. AgentPlan.addSkills now merges three LinkedHashSets but the test only asserts on getPaths(). Add cases for overlapping urls, overlapping classpathResources, and a mixed three-source case asserting all three end up in the merged result.

[wzhero1]

API asymmetry. Java fromClasspath(String...) is varargs; Python from_package(package, resource) is single-pair, so two packages require two @skills functions.

Make it symmetric: def from_package(cls, *pairs: Tuple[str, str]). See also the cross-language drop thread on the packages field.

[wenjin272]

Hi, @wzhero1, Ty for your thorough review.

I fixed some design issues and bugs, including

• Skills rewritten as a single discriminated sources: List<SkillSourceSpec(scheme, params)> + SkillSourceRegistry. Uniform Java/Python wire format.
• URL / Classpath / Package / FileSystem repos are now sibling SkillRepository impls composing a shared SkillDirectoryReader. No more inheritance-as-composition.
• SkillOrigin(scheme, location) attached to every AgentSkill at registration.
• SkillRepository extends AutoCloseable; SkillManager.close() cascades through ResourceContextImpl on operator close (failover-safe).
• Duplicate skill names now LOG.warn with both origins; last-write-wins kept.
• All matching classpath JARs merged via extractClasspathFromJars; same-path collisions WARN.
• Single-pass zip extract; JAR_URL_PREFIX_LEN named constant.

The remaining issues related to robustness and security will be iterated on in subsequent phases.

[wzhero1]

Round-2 follow-up review against the new commits (485573a → d6d58d8). Per-comment scope: lifecycle / cross-language consistency / API hygiene gaps left after the prior round, plus one architectural smell carried into the registry refactor. None of these block merge individually; the constructor-leak (#5) and the non-deterministic merge order (#7) are the most worth addressing this PR.

[wzhero1]

Java/Python serialization surface for origin is asymmetric.

Python's AgentSkill extends BaseModel, so origin is in default model_dump() output. Java's AgentSkill is a POJO with no Jackson annotation — any future debug/checkpoint/Pemja serialization path silently drops it on Java while Python emits it. Same class of cross-language drop A3 fixed at the Skills envelope.

Add @JsonProperty("origin") + @JsonInclude(NON_NULL). Near-zero cost now; cost grows once AgentSkill travels on the wire.

[wenjin272]

Took a closer look and decided to skip this for now. Grep'd every AgentSkill reference in the repo (Java + Python) — no serialization path uses it today, and Jackson's
default bean introspection would emit getOrigin() as "origin" automatically even without the annotation, so the "silently drops" failure mode doesn't actually fire under
the project's default ObjectMapper.

Also worth noting: I ran Python's model_dump() and it emits "origin": null by default (pydantic v2 doesn't drop None fields), so @JsonInclude(NON_NULL) would have made
the two sides byte-asymmetric. I'd rather design the schema (NON_NULL vs null, resources handling, @JsonCreator) when there's an actual wire path with concrete
requirements. Happy to revisit if you have a specific path in mind.

[wenjin272]

Hi, @wzhero1. Ty for the careful review.

I have updated and addressed the second round comments:

• owned rename + 5× delegating boilerplate: the three repos collapsed into AbstractMaterializedSkillRepository (Java) / MaterializedSkillRepository (Python); the
  field becomes materialization on the base. Python's PackageSkillRepository simplified to own its tempdir (no more as_file ctx + atexit two-step). — 68dc94b

• originOf scheme ladder: moved onto SkillSourceHandler.describeLocation; built-ins register both functions together. Dead case "package" is gone. — 85caed1

• Partial-load repo leak + close() swallow: constructor closes already-registered repos before propagating; close cascade aligned with ResourceCache.close()
  (first-exception rethrow + suppressed). — 47360d2

• @Skills merge non-determinism: addSkills sorts keys lexically before merging; mixed-scheme test re-asserts as List to lock the order. — dcab835

• AgentSkill JSON asymmetry: not taking — see inline reply.

Six of seven taken; cross-language symmetry preserved on every fix.

[wzhero1]

This branch is not base64-encoding — binary resources are silently corrupted.

resources[relative_path] = f"base64: {file_path.read_bytes()}"

The f-string returns the bytes repr (literal "base64: b'\\x89PNG...'"), not encoded data. base64.b64decode(value.removeprefix("base64:")) will raise on any real payload. Java sibling uses Base64.getEncoder().encodeToString(...), so the cross-language contract diverges silently and every binary resource read via AgentSkill.get_resource() on Python is unusable.

Fix: "base64:" + base64.b64encode(file_path.read_bytes()).decode("ascii"). Worth adding a PNG round-trip test through both languages — none exists today.

[wzhero1]

Partial-load leak: a failed source leaves earlier-opened repos uncleaned.

Java's loadAll() now wraps each open() in try/catch and closes already-opened repos on failure (round-2 follow-up). Python here still just re-raises — if source #3 of 5 fails, sources #1 and #2 keep their extracted temp dirs and atexit handlers for the interpreter's lifetime. In a long-lived TaskManager Python worker every failed reload leaks.

Same try/finally + first-exception pattern as the Java side: collect opened repos in a list, close them on the failure path before re-raising.

[wzhero1]

Duplicate skill name overwrites the repo-index entry; the displaced repo is orphaned from closeRepos(). Bug exists in both languages.

repos.put(skillName, repo);   // overwrites previous repo ref

closeRepos() iterates repos.values(), so when two sources contribute skill "foo" the displaced repo is no longer reachable — its temp dir survives until JVM exit, only the shutdown hook ever fires. The recently-added close-cascade plumbing can't help because the orphan is never visited. Python mirrors the same shape at python/.../skill_manager.py:177.

Fix: track opened repos in a separate List<SkillRepository> and iterate that in closeRepos() (dedup with IdentityHashMap).

[wenjin272]

Hi, @wzhero1, thanks for round three. All three taken:

• Python base64 silently corrupting binary resources: spot on — f-string was inlining the bytes repr, not real base64. Switched to
  base64.b64encode(...).decode("ascii"). The single-line startsWith("base64: ") assertion is exactly how this hid for a cycle; both Java and Python tests now b64-decode the
  value and assert the round-trip against a PNG signature + non-UTF-8 byte sequence. — e795648
• Python partial-load leak: Python _load_skills now closes already-opened repos before re-raising, mirroring the Java fix from round two. — aa7c978
• Duplicate-name repo orphan in close(): confirmed on both sides — the skill-name → repo map drops the displaced repo's reference, so iterating it on close orphaned the
  earlier repo. Added an openedRepos/_opened_repos list (identity / id() dedup) and changed close to iterate that. New test on each side: two sources contributing the
  same skill name, both repos must be closed. — 8b10d3a

[wzhero1]

approved