工具权限、审批与沙箱契约 深度解析

15@dataclass(frozen=True)
16class ToolMeta:
17    """Metadata for a tool instance.
18
19    Attributes:
20        is_read_only: Tool only reads data, never modifies state.
21        is_destructive: Tool can cause irreversible changes (file deletion, etc.).
22        concurrency_safe: Safe to run in parallel with other tools.
23        timeout: Maximum execution time in seconds.
24        category: Logical grouping (web, file, compute, search, vision).
25        max_result_chars: Max characters to keep inline. Overflow saved to disk.
26            0 means no limit (default for most tools).
27        result_is_ranked: The result is ordered by RELEVANCE, best first, so a
28            contiguous head is worth more than a head-plus-tail split — the tail
29            of a ranked list is its worst entries. False (the default) means the
30            result is sequential: a transcript, a document, a file listing, an
31            exec log, where the end carries the verdict and is worth keeping.
32            Read by ``_overflow`` when TOOL_RESULT_TRUNCATION=auto.
33    """
34
35    is_read_only: bool = True
36    is_destructive: bool = False
37    concurrency_safe: bool = True
38    timeout: int = 45
39    category: str = ""
40    max_result_chars: int = 0
41    result_is_ranked: bool = False
42

431        # Rewrite an absolute cwd-internal path to relative so file tools take
432        # the fast local path (abs paths hit a ~50s sandbox bootstrap). The
433        # rewritten args are what actually executes (via rewrite_args).
434        rewritten = localize_path_args(name, args, self.cwd)
435        eff_args = rewritten if rewritten is not None else args
436
437        # Plan mode: ``exit_plan_mode`` is a human-approval gate that unlocks
438        # edits; every other mutating call is blocked until then.
439        if name == "exit_plan_mode":
440            self._render_activity_call(name, eff_args, call_id=call_id)
441            intervention = await self._review_plan(
442                str(eff_args.get("plan", "")).strip(),
443            )
444            if intervention.skip_with_result is not None:
445                self._activity_synthetic[-1] = True
446            return intervention
447        if (self.plan_state is not None and self.plan_state.active
448                and is_mutating_tool(name, eff_args)):
449            self._render_tool_call(
450                name, eff_args, call_id=call_id, risk_reason="plan mode",
451            )
452            self.r.note(f"✗ plan mode: '{name}' is disabled until the plan is approved")
453            return self._skip_tool(
454                f"[Plan mode is active — '{name}' is disabled. Investigate with "
455                "read/search tools, then call exit_plan_mode(plan=...) to get the "
456                "user's approval before editing.]"
457            )
458
459        risk = assess_with_rules(
460            name, eff_args, self.cwd, self.rules,
461            auto_for_me=getattr(self.approver, "auto_for_me", False),
462        )
463        self._render_tool_call(
464            name, eff_args, call_id=call_id,
465            risk_reason=(risk.danger or ("" if risk.level == RISK_SAFE else risk.reason)),
466            danger=bool(risk.danger),
467        )
468
469        # Preview of the proposed action, shown *inside* the approval modal so
470        # the user can see exactly what they're approving (diff / command).
471        preview = ""
472        preview_kind = ""
473        # Read-before-edit: don't let the model overwrite/edit an existing file
474        # it hasn't read this session, or one changed on disk since it read it.
475        if name in _DIFF_TOOLS:
476            guard = fsguard.check_can_edit(
477                str(eff_args.get("path") or eff_args.get("file_path") or ""), self.cwd,
478            )
479            if guard:
480                self.r.note(f"✗ {name} blocked: {guard}")
481                return self._skip_tool(f"[{guard}]")
482            # Show exactly what will change on disk before asking for approval.
483            diff = unified_diff(name, eff_args, self.cwd)
484            if diff:
485                self.r.diff_preview(diff, stats=change_stats(name, eff_args, self.cwd))
486                preview, preview_kind = diff, "diff"
487        elif name == "bash":
488            cmd = str(eff_args.get("command", "")).strip()
489            if cmd:
490                preview, preview_kind = cmd, "command"
491
492        if risk.level == RISK_DENY:
493            self.r.note(f"✗ blocked: {risk.reason}")
494            return self._skip_tool(
495                f"[blocked by safety policy: {risk.reason}]",
496            )
497        if risk.level != RISK_SAFE:  # confirm: ask the human
498            decision = await self.approver.confirm(
499                name, risk.target, risk.reason, dangerous=risk.danger,
500                preview=preview, preview_kind=preview_kind,
501            )
502            if not decision.approved:
503                # Human-in-the-loop: a typed instruction redirects the agent;
504                # it's fed back as the (declined) call's result so the model
505                # adapts on the next turn instead of just retrying blindly.
506                if decision.feedback:
507                    self.r.note(f"↳ redirecting {name}: {decision.feedback}")
508                    return self._skip_tool(
509                        f"[The user declined to run this {name} call. "
510                        f"Follow their instruction instead: {decision.feedback}]"
511                    )
512                # Plain reject → stop the task and hand control back (Claude /
513                # kimi behaviour). Re-trying a rejected action just makes the
514                # model ramble; ``[e] redirect`` is how the user keeps it going.
515                self._abort_reason = "user_rejected"
516                self.r.note(f"✗ rejected {name} — stopping task (use [e] to redirect instead)")
517                return self._skip_tool(
518                    f"[user rejected this {name} call — task stopped]",
519                )
520            # 'Always allow this command' → persist a rule so identical calls
521            # auto-approve next time (danger/deny still override it).
522            if decision.remember and self.rules is not None:
523                self.r.note(f"✓ always allowing: {self.rules.add_allow(name, eff_args)}")
524
525        # Allowed (safe or approved) → snapshot a mutating op before it runs so
526        # the change is diffable + revertable, then let it execute.
527        if self.journal is not None and name in MUTATING_TOOLS:
528            p = eff_args.get("path") or eff_args.get("file_path")
529            if isinstance(p, str) and p:
530                self.journal.record_before(p)
531        elif self.journal is not None and is_mutating_tool(name, eff_args):
532            # Awaited before the call executes, so the baseline is always in
533            # place first — including for a sibling that reuses it.
534            await self._begin_journal_scan()
535        return ToolCallIntervention(rewrite_args=rewritten) if rewritten is not None else None

167class AgentTeamAssignmentSpec(AssignmentSpec):
168    """One assignment. Carrying ``output_paths`` is what authorizes it to write
169    those ``/outputs`` paths; without them the task runs workspace-only."""
170
171    model_config = ConfigDict(extra="forbid")
172
173    publish: StrictBool | None = Field(
174        default=None,
175        # Marked deprecated in the advertised schema only. Pydantic's own
176        # ``deprecated=True`` warns on every attribute read, and
177        # ``validate_publish_contract`` below reads it twice per assignment --
178        # that is a DeprecationWarning per validated task in every run.
179        json_schema_extra={"deprecated": True},
180        description=(
181            "Deprecated, omit it. output_paths alone grants publication. If "
182            "sent it must agree with the manifest."
183        ),
184    )
185    output_paths: list[StrictStr] = Field(
186        default_factory=list,
187        description=(
188            "Exact absolute final file paths under /outputs. Supplying them IS "
189            "the publication grant, so set them on exactly one final-publisher "
190            "assignment and omit them on every other. Pass a JSON array even "
191            "for one path."
192        ),
193    )
194    replace_manifest: StrictBool = Field(
195        default=False,
196        description=(
197            "Set true only on a follow-up to the existing publisher when the "
198            "required final output formats genuinely changed."
199        ),
200    )
201
202    @property
203    def can_publish(self) -> bool:
204        """Whether this assignment may write ``/outputs``.
205
206        Derived from the manifest alone: a path list is the grant.
207        """
208        return bool(self.output_paths)
209
210    @field_validator("publish", mode="before")
211    @classmethod
212    def normalise_publish_boolean(cls, value: Any) -> Any:
213        """Accept common JSON boolean strings in the structured field."""
214        if isinstance(value, str):
215            normalised = value.strip().lower()
216            if normalised in {"false", "true"}:
217                return normalised == "true"
218        return value
219
220    @field_validator("output_paths", mode="before")