T02 · Agent Memory Poisoning
- Location
scripts/context_guardian.py:245- Finding
Persistent agent-memory poisoning through untrusted durable state and summaries
- Content
View full analysis
"TaskState": plan = [TaskStep(**step) for step in payload.get("plan", [])] artifacts = [TaskArtifact(**artifact) for artifact in payload.get("artifacts", [])] last_action_payload = payload.get("last_action") last_action = LastAction(**last_action_payload) if last_action_payload else None return cls( task_id=payload["task_id"], session_id=payload["session_id"], goal=payload["goal"], status=payload.get("status", "idle"), current_phase=payload.get("current_phase", "initial"), plan=plan, completed_steps=list(payload.get("completed_steps", [])), open_issues=list(payload.get("open_issues", [])), decisions=list(payload.get("decisions", [])), constraints=list(payload.get("constraints", [])), important_facts=list(payload.get("important_facts", [])), artifacts=artifacts, last_action=last_action, next_action=payload.get("next_action", ""), recovery_instructions=list(payload.get("recovery_instructions", [])), state_confidence=float(payload.get("state_confidence", 0.0)), updated_at=payload.get("updated_at", iso_now()), ) ``` ```python def latest_task_state(self) -> TaskState | None: if not self.task_state_json.exists(): return None payload = json.loads(self.task_state_json.read_text(encoding="utf-8")) return TaskState.from_dict(payload) def latest_summary(self) -> str | None: if not self.summaries_dir.exists(): return None summaries = sorted(self.summaries_dir.glob("*.md")) if not summaries: return None return summaries[-1] ...[truncated 3999 chars]- Remediation
View remediation
