feat/app-feed

README.md

@@ -169,6 +169,21 @@ python -m webwright.run.cli \
     -o outputs/default
 ```
 
+Use the YAML-only app/feed prompt overlay for multisite local PWA information
+feeds:
+
+```bash
+python -m webwright.run.cli \
+    -c base.yaml -c model_openai.yaml -c app_feed.yaml \
+    -t "Build a local feed app that compares current Apple Mac laptop prices across Apple, Best Buy, and B&H. Use browser evidence from the product/listing pages, and show each model with source, price, configuration, availability if visible, original link, and visible source errors." \
+    --task-id mac_price_feed \
+    -o outputs/default
+```
+
+This path is driven by `src/webwright/config/app_feed.yaml`; the app/feed
+contract is embedded in that YAML, and the agent only reads the packaged
+`webwright/reference/app_feed_template.py` file as the code skeleton.
+
 ### 🚩 Flags
 
 | Flag | Description |

pyproject.toml

@@ -26,4 +26,4 @@ webwright = "webwright.run.cli:app"
 where = ["src"]
 
 [tool.setuptools.package-data]
-webwright = ["config/*.yaml", "config/**/*.yaml"]
+webwright = ["config/*.yaml", "config/**/*.yaml", "reference/*.py"]

src/webwright/agents/default.py

@@ -39,6 +39,9 @@ class AgentConfig(BaseModel):
     require_self_reflection_success: bool = False
     summary_every_n_steps: int = 0
     summary_user_prompt: str = DEFAULT_SUMMARY_USER_PROMPT
+    compact_observation_history: bool = False
+    recent_observation_limit: int = 4
+    compacted_observation_chars: int = 1200
     output_path: Path | None = None
 
 
@@ -305,6 +308,55 @@ def _compact_history(self) -> None:
         )
         self.messages = [system_message, summary_message]
 
+    def _observation_summary_for_context(self, observation: dict[str, Any]) -> str:
+        lines = [
+            "Observation compacted. Full command and output remain on disk.",
+            f"Status: {'ok' if observation.get('success') else 'error'}",
+            f"Return code: {observation.get('returncode', '')}",
+        ]
+        for label, key in (
+            ("Command file", "command_path"),
+            ("Step log", "log_path"),
+            ("Final script", "final_script_path"),
+            ("Latest screenshot", "screenshot_path"),
+        ):
+            value = observation.get(key)
+            if value:
+                lines.append(f"{label}: {value}")
+        exception = str(observation.get("exception") or "").strip()
+        if exception:
+            lines.append(f"Exception: {exception[: self.config.compacted_observation_chars]}")
+        files = observation.get("workspace_files")
+        if isinstance(files, list) and files:
+            lines.append("Recent files: " + ", ".join(str(item) for item in files[:10]))
+        text = "\n".join(lines)
+        if len(text) <= self.config.compacted_observation_chars:
+            return text
+        omitted = len(text) - self.config.compacted_observation_chars
+        return f"{text[: self.config.compacted_observation_chars]}\n... [{omitted} characters omitted]"
+
+    def _compact_old_observations(self) -> None:
+        if not self.config.compact_observation_history:
+            return
+        observation_indices = [
+            index
+            for index, message in enumerate(self.messages)
+            if message.get("role") == "user" and isinstance(message.get("extra", {}).get("observation"), dict)
+        ]
+        keep = max(self.config.recent_observation_limit, 0)
+        compact_indices = observation_indices if keep == 0 else observation_indices[:-keep]
+        for index in compact_indices:
+            message = self.messages[index]
+            extra = dict(message.get("extra", {}))
+            if extra.get("observation_compacted"):
+                continue
+            observation = extra.get("observation")
+            if not isinstance(observation, dict):
+                continue
+            extra["observation_compacted"] = True
+            message["content"] = self._observation_summary_for_context(observation)
+            message["extra"] = extra
+
     def run(self, task: str = "", **kwargs) -> dict[str, Any]:
         self.extra_template_vars |= {"task": task, **kwargs}
         self.messages = []
@@ -359,6 +411,7 @@ def query(self) -> dict[str, Any]:
                     extra={"exit_status": "LimitsExceeded", "submission": ""},
                 )
             )
+        self._compact_old_observations()
         message = self.model.query(self.messages)
         self.n_calls += 1
         self.add_messages(message)

src/webwright/config/base.yaml

@@ -66,6 +66,7 @@ environment:
   task_metadata_filename: task.json
   final_script_name: final_script.py
   output_truncation_chars: 24000
+  command_preview_chars: 2000
   final_script_preview_chars: 4000
   recent_files_limit: 40
   env:

src/webwright/environments/local_workspace.py

@@ -3,13 +3,15 @@
 import json
 import os
 import re
+import shutil
 import subprocess
+import sys
 from pathlib import Path
 from typing import Any
 
 from pydantic import BaseModel, Field
 
-_EXPORT_RE = re.compile(r"^export\s+([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
+_ENV_ASSIGNMENT_RE = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=(.*)$")
 _IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp"}
 
 
@@ -40,6 +42,7 @@ class LocalWorkspaceEnvironmentConfig(BaseModel):
     task_metadata_filename: str = "task.json"
     final_script_name: str = "final_script.py"
     output_truncation_chars: int = 12000
+    command_preview_chars: int = 2000
     final_script_preview_chars: int = 4000
     recent_files_limit: int = 40
 
@@ -64,7 +67,7 @@ def _load_credential_env(self, path: Path | None) -> dict[str, str]:
             line = raw_line.strip()
             if not line or line.startswith("#"):
                 continue
-            match = _EXPORT_RE.match(line)
+            match = _ENV_ASSIGNMENT_RE.match(line)
             if match is None:
                 continue
             key, raw_value = match.groups()
@@ -74,6 +77,27 @@ def _load_credential_env(self, path: Path | None) -> dict[str, str]:
             parsed[key] = value
         return parsed
 
+    def _python_env(self) -> dict[str, str]:
+        """Expose the runner interpreter to agent-authored shell commands.
+
+        Webwright requires Python >=3.10. If the CLI is launched from an activated
+        environment such as ``conda activate metachain``, subprocess commands
+        should use that same interpreter instead of falling back to a system/base
+        ``python`` that may be too old.
+        """
+
+        configured = os.environ.get("WEBWRIGHT_PYTHON") or sys.executable
+        resolved = shutil.which(configured) if not Path(configured).is_absolute() else configured
+        python_path = str(Path(resolved or configured).resolve())
+        python_dir = str(Path(python_path).parent)
+        current_path = os.environ.get("PATH", "")
+        path = python_dir if not current_path else f"{python_dir}{os.pathsep}{current_path}"
+        return {
+            "WEBWRIGHT_PYTHON": python_path,
+            "PYTHON": python_path,
+            "PATH": path,
+        }
+
     def _workspace_dir(self) -> Path:
         return self.config.output_dir.resolve()
 
@@ -178,10 +202,10 @@ def execute(self, action: dict[str, Any], cwd: str = "") -> dict[str, Any]:
         command = str(
             action.get("command") or action.get("bash_command") or action.get("python_code") or ""
         ).strip()
-        self._persist_step_command(command)
+        step_path = self._persist_step_command(command)
         resolved_cwd = self._resolve_cwd(cwd)
 
-        command_env = os.environ | self._credential_env | self._browser_env() | self.config.env | {
+        command_env = os.environ | self._credential_env | self._python_env() | self._browser_env() | self.config.env | {
             "WORKSPACE_DIR": str(self._workspace_dir()),
             "OM2W_TASK_JSON": str(self._task_metadata_path()),
             "FINAL_SCRIPT_PATH": str(self._final_script_path()),
@@ -214,6 +238,7 @@ def execute(self, action: dict[str, Any], cwd: str = "") -> dict[str, Any]:
         log_path = self._write_step_log(output)
         observation = self._capture_observation(
             command=command,
+            step_path=step_path,
             cwd=resolved_cwd,
             output=output,
             returncode=returncode,
@@ -231,6 +256,7 @@ def _capture_observation(
         self,
         *,
         command: str,
+        step_path: Path,
         cwd: Path,
         output: str,
         returncode: int,
@@ -252,7 +278,8 @@ def _capture_observation(
         return {
             "success": returncode == 0 and not exception_info,
             "exception": exception_info,
-            "command": command,
+            "command": self._truncate(command, self.config.command_preview_chars),
+            "command_path": str(step_path),
             "returncode": returncode,
             "workspace_dir": str(workspace_dir),
             "cwd": str(cwd),

src/webwright/models/base.py

@@ -209,6 +209,7 @@ class BaseModelConfig(PydanticBaseModel):
     request_timeout_seconds: int = 120
     error_log_path: Path | None = None
     observation_template: OptStr = DEFAULT_OBSERVATION_TEMPLATE
+    observation_truncation_chars: int = 12000
     format_error_template: OptStr = DEFAULT_FORMAT_ERROR_TEMPLATE
     attach_observation_screenshot: bool = True
     action_field: str = "bash_command"
@@ -384,6 +385,10 @@ def format_observation_messages(
                 observation=observation,
                 **(template_vars or {}),
             )
+            limit = self.config.observation_truncation_chars
+            if limit > 0 and len(content) > limit:
+                omitted = len(content) - limit
+                content = f"{content[:limit]}\n\n... [{omitted} observation characters omitted]"
 
             parts: list[dict[str, Any]] = [text_part(content)]
             screenshot_path = observation.get("screenshot_path")