Wrapper gaps in src/praisonai/praisonai: silent YAML duplicate drops, global-max tool_timeout, and skill nested-path lockout

[MervinPraison]

Scope

Wrapper layer only: src/praisonai/praisonai/. Every finding was located by reading the file at HEAD (branch claude/bold-bohr-kdae6d), reproduced with a runnable snippet, and cross-checked against the philosophy pillars:

• Agent-centric, safe by default, multi-agent safe by default.
• Minimal API, sensible defaults, explicit overrides — silent data loss is not an acceptable default.
• Few-lines-of-code UX for non-developers — a feature that half-works in a common shape violates this.

Not about documentation, tests, coverage, file sizes, or line counts. Three concrete behavioural gaps below, each with reproducer + patch sketch.

---

1. _normalize_yaml_config silently drops duplicate agents/tasks and reduces multi-agent workflows to fewer agents than the user declared

Pillar violated: safe by default; simple to adopt, hard to misuse.

File: src/praisonai/praisonai/agents_generator.py, lines 37–89 (_normalize_yaml_config).

What happens

The normaliser converts list-form agents:/tasks: YAML into dicts using the entry's name as the dict key. Two entries with the same name silently collapse — the later one wins, the earlier one is discarded with no warning, no error, and no visibility in the config validator (which runs after normalisation, so the duplicates are gone by the time _validate_agents_config looks). Tasks referencing an unknown agent are dropped with only a logger.warning and the run proceeds with fewer tasks than the user wrote.

Reproducer (verified on this branch — output shown after the snippet):

from praisonai.agents_generator import _normalize_yaml_config

cfg = {
    "agents": [
        {"name": "writer", "role": "Writer A", "instructions": "A"},
        {"name": "writer", "role": "Writer B", "instructions": "B"},  # duplicate name
    ],
    "tasks": [
        {"name": "t1", "agent": "writer", "description": "t1"},
        {"name": "t1", "agent": "writer", "description": "t1-B"},   # duplicate task name
        {"name": "t2", "agent": "nonexistent", "description": "t2"}, # bad ref, silently dropped
    ],
}
out = _normalize_yaml_config(cfg)
print(list(out["agents"]))                                 # ['writer']            — ONE agent left
print({a: list(r.get("tasks", {})) for a, r in out["roles"].items()})
# {'writer': ['t1']}   — Writer A gone, task t1 (first copy) gone, t2 gone

Observed on claude/bold-bohr-kdae6d:

agents kept: ['writer']
  writer: role=Writer B, tasks=['t1']

Only a WARNING is emitted for the bad-agent-ref task; the two duplicate collisions are completely silent.

Why this is a real gap

• A YAML that declares three agents runs with one, and the run "succeeds" — every stated pillar (agent-centric, multi-agent safe, non-developer-friendly) says this must fail fast or at minimum warn loudly.
• _validate_agents_config (line 906) uses ConfigValidator and even supports PRAISONAI_VALIDATE_STRICT=true, but the strict mode can never see the duplicates because they were already collapsed by _normalize_yaml_config in _load_config (line 964).
• The _normalize_yaml_config return value is also used inside AutoGenerator hand-off paths, so the same drop happens for any callsite that normalises externally-authored YAML.

Fix sketch (src/praisonai/praisonai/agents_generator.py:37–89):

def _normalize_yaml_config(config: dict) -> dict:
    if not isinstance(config, dict):
        return config

    agents = config.get("agents")
    if isinstance(agents, list):
        normalized, duplicates = {}, []
        for i, a in enumerate(agents):
            if not isinstance(a, dict):
                continue
            key = a.get("name") or f"agent_{i}"
            if key in normalized:
                duplicates.append(key)
                # Preserve BOTH by suffixing the collision instead of clobbering.
                key = f"{key}__dup_{i}"
            normalized[key] = a
        if duplicates:
            import os
            msg = f"Duplicate agent name(s) in YAML: {sorted(set(duplicates))}"
            if os.getenv("PRAISONAI_VALIDATE_STRICT", "false").lower() == "true":
                raise ValueError(msg)
            logger.warning("%s — kept both by suffixing keys; rename to silence.", msg)
        config["agents"] = normalized

    # Same pattern for tasks: detect collisions, raise/warn, never silently
    # drop. And for tasks that reference an unknown agent:
    tasks = config.get("tasks")
    if isinstance(tasks, list):
        ...
        for i, task in enumerate(tasks):
            ...
            if not agent_key or agent_key not in bucket:
                if os.getenv("PRAISONAI_VALIDATE_STRICT", "false").lower() == "true":
                    raise ValueError(
                        f"Task {task.get('name', f'task_{i}')!r} references "
                        f"unknown agent {agent_key!r}."
                    )
                logger.warning(...)
                continue
    return config

This is a small change that turns three silent collapses into three loud, actionable errors without breaking any run that was already correct.

---

2. BasicSkillMutator._is_safe_path rejects every nested skill file, silently breaking a documented feature

Pillar violated: safe by default — but "safe" must not mean "unusable"; the check is over-broad and blocks the feature it's meant to protect.

File: src/praisonai/praisonai/tools/skill_manage.py, lines 449–464 (_is_safe_path), and its callers at lines 89–103 (patch), 196–235 (write_file), 237–276 (remove_file).

What happens

_is_safe_path refuses any path containing /:

def _is_safe_path(self, file_path: str) -> bool:
    if not file_path:
        return False
    dangerous_patterns = ["..", "/", "\\", "~"]  # <-- "/" is on the deny-list
    if any(pattern in file_path for pattern in dangerous_patterns):
        return False
    normalized = os.path.normpath(file_path)
    if normalized != file_path or normalized.startswith("/"):
        return False
    return True

Every skill in the ecosystem has a canonical layout with sub-directories — references/, scripts/, assets/ — so an agent trying to persist a real skill via skill_manage("write_file", "python-debugging", "references/palette.md", ...) gets:

❌ Invalid file path 'references/palette.md'. Path traversal not allowed.

Meanwhile, the safe structural check right below it —

try:
    file_to_edit.resolve().relative_to(skill_path.resolve())
except ValueError:
    return f"❌ Invalid file path '{file_path}'. Must be within skill directory."

— already provides real containment (it resolves symlinks and checks the resulting path is inside the skill dir). The /-deny-list is redundant and wrong.

Reproducer (verified — output shown after the snippet):

import tempfile
from praisonai.tools.skill_manage import BasicSkillMutator
with tempfile.TemporaryDirectory() as td:
    m = BasicSkillMutator(skills_dir=td)
    m.create("test-skill", "# hello", propose=False)
    print(m.write_file("test-skill", "notes.md", "ok", propose=False))
    print(m.write_file("test-skill", "references/palette.md", "red", propose=False))
    print(m.patch("test-skill", "hello", "world",
                  file_path="references/palette.md", propose=False))

Observed:

✅ File 'notes.md' written to skill 'test-skill'.
❌ Invalid file path 'references/palette.md'. Path traversal not allowed.
❌ Invalid file path 'references/palette.md'. Path traversal not allowed.

The flat file works; a canonical nested reference file cannot ever be written or patched via this tool.

Fix sketch (src/praisonai/praisonai/tools/skill_manage.py:449–464):

def _is_safe_path(self, file_path: str) -> bool:
    """Reject absolute paths and explicit parent-directory traversal.

    Real containment (symlinks, resolved paths) is enforced by the caller
    with ``file_to_edit.resolve().relative_to(skill_path.resolve())`` — this
    is a fast, syntactic prefilter, not the security boundary.
    """
    if not file_path:
        return False
    if file_path.startswith(("/", "~")) or "\0" in file_path:
        return False
    # Split on either separator so "..\\x" is caught on all OSes.
    parts = file_path.replace("\\", "/").split("/")
    if any(p in ("..", "") for p in parts):
        return False
    return True

Now references/palette.md and scripts/helper.py work, while ../etc/passwd, /etc/passwd, ~/.ssh/id_rsa, and empty segments are still rejected — and the .resolve().relative_to(...) post-check keeps symlink-based escapes closed.

While in the file, also worth addressing

• create() (line 44) checks if skill_path.exists() then calls mkdir(parents=True, exist_ok=True) — replace with mkdir(exist_ok=False) and catch FileExistsError to close the TOCTOU race between two concurrent creates. Same shape in approve() (line 356) and write_file() (line 231).
• _log_action (line 486) does a read-modify-write on .skill_log with no lock — two concurrent skill mutations lose entries. Either append to a JSONL file with open(..., "a") (atomic single-line writes on POSIX) or wrap the RMW in fcntl.flock. The "Multi-agent + async safe by default" pillar is explicit about this.

---

3. Per-agent tool_timeout is coerced to a single process-wide max() — a fast agent silently inherits the slowest agent's timeout

Pillar violated: multi-agent safe by default; explicit overrides.

File: src/praisonai/praisonai/agents_generator.py, lines 743–782 (_build_tools_dict, _resolve_effective_tool_timeout).

What happens

_build_tools_dict wraps the shared tool objects with a single timeout resolved from _resolve_effective_tool_timeout:

def _build_tools_dict(self, config):
    tools_dict = self.tool_resolver.resolve_all_from_yaml(config)
    ...
    effective = self._resolve_effective_tool_timeout(config)   # ONE global number
    if effective and effective > 0:
        tools_dict = {
            name: self._wrap_tool_with_timeout(tool, effective)
            for name, tool in tools_dict.items()
        }
    return tools_dict

def _resolve_effective_tool_timeout(self, config):
    ...
    entities = {**config.get("roles", {}), **config.get("agents", {})}
    timeouts = [
        e.get("tool_timeout") for e in entities.values()
        if isinstance(e, dict)
        and isinstance(e.get("tool_timeout"), (int, float))
        and not isinstance(e.get("tool_timeout"), bool)
    ]
    return float(max(timeouts)) if timeouts else None

The docstring calls max() the "safest default for a shared tool dict". It is the opposite of safe from a multi-agent viewpoint: an agent that declared tool_timeout: 5 because it wants to bail out quickly on a slow tool will now wait 60s because another agent in the same YAML wanted a longer budget for the same tool. There is no per-agent enforcement anywhere in the wrapper timeout stack — the per-agent field is quietly advisory.

Reproducer (verified — output shown after the snippet):

from praisonai.agents_generator import AgentsGenerator

class Gen(AgentsGenerator):
    def __init__(self):
        self.cli_config = {}
        import logging; self.logger = logging.getLogger("t")

gen = Gen()
config = {"roles": {
    "fast_agent": {"role": "A", "goal": "x", "backstory": "y", "tool_timeout": 5},
    "slow_agent": {"role": "B", "goal": "x", "backstory": "y", "tool_timeout": 60},
}}
print("Effective timeout applied to *both*:", gen._resolve_effective_tool_timeout(config), "s")

Effective timeout applied to *both*: 60.0 s

fast_agent asked for 5s, gets 60s. The user has no signal this happened.

Why this matters

• The tool-timeout wrapping architecture (_wrap_with_timeout, per-instance executor, pool-recycling on leaks) is clearly designed to be multi-agent safe (docstring even says so). The scoping is right — but the value is collapsed above it, so every downside of a global timeout returns.
• The 3-way surface promises --tool-timeout on the CLI, tool_timeout in YAML per role/agent, and tool_timeout= on the Python Agent constructor. Only the CLI form and a single-agent YAML actually work as documented.

Fix sketch — resolve per-agent, wrap per-agent, key the tools dict by (agent, tool):

def _build_tools_dict(self, config):
    base_tools = self.tool_resolver.resolve_all_from_yaml(config)
    for tool_class in self.tools:
        if isinstance(tool_class, type):
            try:
                base_tools[tool_class.__name__] = tool_class()
            except (TypeError, ValueError, RuntimeError) as e:
                self.logger.warning("Failed to instantiate %s: %s", tool_class.__name__, e)

    # Per-agent timeout, applied to a per-agent view of the tool dict so a
    # shared tool object stays untouched and each agent sees its own wrapper.
    entities = {**config.get("roles", {}), **config.get("agents", {})}
    per_agent_tools: Dict[str, Dict[str, Any]] = {}
    cli_timeout = (self.cli_config or {}).get("tool_timeout")
    cli_timeout = float(cli_timeout) if isinstance(cli_timeout, (int, float)) and not isinstance(cli_timeout, bool) else None

    for agent_name, entity in entities.items():
        if not isinstance(entity, dict):
            continue
        raw = entity.get("tool_timeout")
        agent_to = cli_timeout if cli_timeout is not None else (
            float(raw) if isinstance(raw, (int, float)) and not isinstance(raw, bool) else None
        )
        if agent_to and agent_to > 0:
            per_agent_tools[agent_name] = {
                name: self._wrap_tool_with_timeout(tool, agent_to)
                for name, tool in base_tools.items()
            }
        else:
            per_agent_tools[agent_name] = dict(base_tools)  # no wrap

    # Return the base dict for adapters that don't consume the per-agent form,
    # plus expose the per-agent map so adapters that CAN honour it will.
    return {"__base__": base_tools, "__per_agent__": per_agent_tools}

Adapters that accept a plain tools_dict keep working with __base__; the PraisonAIAdapter (which knows which agent is running each turn) can look up __per_agent__[agent_name] and get real per-agent enforcement. _wrap_with_timeout's idempotency marker is already keyed by the executor identity, so wrapping the same underlying method for two different agents produces two distinct wrappers as intended.

If the adapter surface change is too large for a first pass, the minimum fix is to switch from max() to min() and log every agent whose declared value was overridden:

tightest = float(min(timeouts))
for name, e in entities.items():
    v = e.get("tool_timeout") if isinstance(e, dict) else None
    if isinstance(v, (int, float)) and not isinstance(v, bool) and v != tightest:
        self.logger.warning(
            "Agent %r declared tool_timeout=%s but a shared tool dict "
            "forces the tightest value %ss for the whole run.",
            name, v, tightest,
        )
return tightest

Silent max() is the worst choice of the three; min() + a warning at least aligns with the "safe by default" pillar.

---

Priority

1. Silent duplicate collapse — highest blast radius, easiest to hit, hardest to notice; a multi-agent YAML looking correct produces a single-agent run.
2. Per-agent tool_timeout collapsed to global max() — breaks a documented per-agent knob and directly contradicts "multi-agent safe by default".
3. Skill nested-path rejection — narrower blast radius but blocks a documented agent-centric workflow outright and is trivial to fix.

All three findings are reproduced by the snippets above against this branch's tree (src/praisonai/praisonai/agents_generator.py, src/praisonai/praisonai/tools/skill_manage.py).

[praisonai-triage-agent]

Claude finished @MervinPraison's task —— View job • claude/issue-3175-20260719-0817 • [Create PR ➔](https://github.com/MervinPraison/PraisonAI/compare/main...claude/issue-3175-20260719-0817?quick_pull=1&title=Issue%20%233175%3A%20Changes%20from%20Claude&body=This%20PR%20addresses%20issue%20%233175%0A%0AGenerated%20with%20%5BClaude%20Code%5D(https%3A%2F%2Fclaude.ai%2Fcode)

---

Done — PR #3176 created ✅

Fixed all three wrapper-layer gaps from #3175 with minimal, backward-compatible changes.

• Read AGENTS.md & architecture rules
• Review affected files
• Fix 1 — Silent YAML duplicate collapse
• Fix 2 — Skill nested-path lockout
• Fix 3 — Per-agent tool_timeout collapse
• Test & verify (54 tests pass)
• Commit, push, create PR

What changed

1. _normalize_yaml_config — agents_generator.py:37
Duplicate agent/task names and unknown task→agent refs now raise under PRAISONAI_VALIDATE_STRICT=true or warn loudly and preserve both entries by suffixing (name__dup_i), instead of silently dropping. Correct YAML is unaffected.

2. _resolve_effective_tool_timeout — agents_generator.py:799
Switched from max() to the tightest (min) declared value — the safe-by-default choice — and warns on every overridden agent. A fast agent (5s) no longer silently inherits a slow agent's 60s budget.

Used the minimal min()+warning fix rather than the per-agent tools-dict restructure, since that new data shape has no live adapter consumer (scope creep per AGENTS.md).

3. _is_safe_path — tools/skill_manage.py:449
Now allows nested skill files (references/, scripts/, assets/) while still rejecting absolute paths, ~, null bytes and .. segments. Real containment stays enforced by each caller's resolve().relative_to() check.

Verification

• Issue reproducers: duplicates preserved+warned; fast_agent→5.0s; nested files write, ../etc/passwd//etc/passwd/~/.ssh/id_rsa rejected.
• test_tool_timeout_enforcement.py: 10 passed (updated test_effective_timeout_uses_tightest_declared_role for max→min).
• skill/normalize/yaml unit tests: 44 passed.

PR: #3176

---

Job run · Branch