DietPi Remote Agent Prototype

0. Implementation Status (updated 2026-07-08)

COMPLETE & DEPLOYED This design has been implemented. Sections 1–11 below are the original handoff design, kept as reference. The authoritative usage documentation is now the project's own README.md.

Deltas from this design

Quick usage

cd ~/dietpi_remote_agent
./publish_command.sh health_check                                  # prove it's alive
./publish_command.sh claude_sdk_run "Check disk usage on this box" # ask Frita
./publish_command.sh make_run                                      # make + run configured C++ project
./publish_command.sh --remote health_check                         # publish to the website instead

tail -f logs/agent.log                                             # audit trail (no reply channel)
python3 -m json.tool data/agent_state.json                         # execution history
systemctl --user restart dietpi-agent                              # after config/code changes

1. Clean Team Prompt

Use this as the main prompt for the implementation agent.

You are the implementation agent for the DietPi Remote Agent prototype.

Goal:
Build a Python-based agent that runs on the remote DietPi system in Florida. For this prototype, do not use SSH. The central Agent Orchestrator will publish command files to the public website path:

  americansjewelry.com/public_html/commands/dietpi/

The DietPi agent must poll the corresponding web URL for new command files, decide whether each command is recent enough to execute, and then execute only approved local actions.

Important prototype constraints:
1. This is a one-way command channel.
2. The DietPi agent should not require an SSH connection.
3. The DietPi agent cannot rely on sending a reply back to the orchestrator.
4. The agent must ignore stale command files.
5. The agent must not execute arbitrary shell commands from the website.
6. Only allowlisted commands may run.
7. All important decisions must be logged locally on the DietPi machine.

Command file naming format:
Each command is represented by a text file. The command name must be snake_case and must include the suggested execution time in the filename.

Example:
  make_run_12_00_00_PM.txt

Meaning:
  Command name: make_run
  Suggested execution time: 12:00:00 PM

Freshness rule:
The DietPi agent must only execute commands that are still recent when discovered.

Example:
If the current DietPi system time is 12:00:01 PM and the queue contains:

  make_run_12_00_00_PM.txt

then the agent must treat that command as stale and ignore it. The agent must assume that commands are intended for their exact suggested execution time, not for later execution.

Implementation expectation:
Python is already installed on the DietPi machine. Build the prototype in Python.

Polling source:
Use the web directory listing or an index file served from:

  https://americansjewelry.com/commands/dietpi/

If HTTPS is not available in the prototype environment, document that limitation and make the base URL configurable. Do not hard-code the URL inside business logic.

Required behavior:
1. Poll the command source on a configurable interval.
2. Discover command files ending in .txt.
3. Parse command filenames into a structured command object.
4. Validate command name, scheduled time, and freshness.
5. Ignore stale, malformed, duplicate, or unknown commands.
6. Execute only approved commands through command handler classes.
7. Log every discovered command and every decision: executed, ignored, stale, malformed, duplicate, unknown, or failed.
8. Keep local state so the same command file is not executed more than once during the agent process lifetime.
9. Provide a clean directory structure with interfaces separated from implementations.
10. Include unit tests for parsing, freshness detection, allowlist validation, duplicate detection, and command dispatch.

Prototype command handlers:
Start with these allowlisted command handlers:

  make_run
    Build and run the configured local C++ project.

  health_check
    Write a local health-check log entry proving the agent is alive.

Do not implement broad remote shell execution. Add new command handlers only through explicit allowlist registration.

Directory guidance:
Place the solution in a dedicated project directory on the DietPi system. Use a clean, testable structure similar to this:

  dietpi_remote_agent/
    README.md
    pyproject.toml
    .env.example
    src/
      dietpi_agent/
        __init__.py
        main.py
        config.py
        models.py
        interfaces.py
        parser.py
        freshness.py
        registry.py
        poller.py
        dispatcher.py
        state.py
        logging_config.py
        handlers/
          __init__.py
          make_run.py
          health_check.py
        infrastructure/
          __init__.py
          http_command_source.py
          local_state_store.py
          subprocess_runner.py
    tests/
      test_command_filename_parser.py
      test_freshness_policy.py
      test_command_registry.py
      test_dispatcher.py
      test_duplicate_detection.py
      test_http_command_source.py

Keep source code under src/dietpi_agent. Keep tests under tests. Keep environment-specific paths in .env or config.py. Do not scatter scripts across unrelated folders.

Configuration must include:
- command_source_url
- poll_interval_seconds
- freshness_window_seconds
- local_project_directory
- make_command
- run_command
- log_file_path
- state_file_path

Design principles:
- Use GoF Command Pattern for executable command handlers.
- Use Strategy Pattern for freshness validation.
- Use Factory or Registry Pattern for command handler lookup.
- Use Repository Pattern for local command state.
- Use Adapter Pattern for the HTTP command source.
- Program to interfaces, not concrete implementations.
- Keep parsing, validation, polling, dispatching, and execution separated.

Acceptance criteria:
1. The agent starts from a single entry point: python -m dietpi_agent.main
2. The agent polls the configured URL.
3. It parses filenames like make_run_12_00_00_PM.txt correctly.
4. It ignores stale commands.
5. It refuses unknown commands.
6. It refuses malformed filenames.
7. It does not run the same command twice.
8. It can execute make_run through an allowlisted handler.
9. Unit tests cover all critical behavior.
10. README.md explains installation, configuration, running, and testing.

2. Important Clarifications for the Team

3. Proposed Directory Structure

The implementation agent should be allowed to choose final paths, but the project should stay organized around clear boundaries.

dietpi_remote_agent/
  README.md
  pyproject.toml
  .env.example
  src/
    dietpi_agent/
      __init__.py
      main.py
      config.py
      models.py
      interfaces.py
      parser.py
      freshness.py
      registry.py
      poller.py
      dispatcher.py
      state.py
      logging_config.py
      handlers/
        __init__.py
        make_run.py
        health_check.py
      infrastructure/
        __init__.py
        http_command_source.py
        local_state_store.py
        subprocess_runner.py
  tests/
    test_command_filename_parser.py
    test_freshness_policy.py
    test_command_registry.py
    test_dispatcher.py
    test_duplicate_detection.py
    test_http_command_source.py

4. Command File Contract

The filename is treated as a command envelope.

5. GoF Design Pattern Mapping

6. Pydantic Models and Interface Contracts

These contracts are intended to guide implementation. They keep the system modular, testable, and easy to extend.

from __future__ import annotations

from datetime import datetime, time
from enum import Enum
from pathlib import Path
from typing import Iterable, Optional, Protocol, Sequence

from pydantic import AnyHttpUrl, BaseModel, Field, PositiveInt, field_validator


class CommandDecision(str, Enum):
    EXECUTE = "execute"
    IGNORE_STALE = "ignore_stale"
    IGNORE_DUPLICATE = "ignore_duplicate"
    IGNORE_MALFORMED = "ignore_malformed"
    IGNORE_UNKNOWN = "ignore_unknown"
    FAILED = "failed"


class ExecutionStatus(str, Enum):
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    SKIPPED = "skipped"


class AgentSettings(BaseModel):
    command_source_url: AnyHttpUrl
    poll_interval_seconds: PositiveInt = 1
    freshness_window_seconds: int = Field(default=0, ge=0, le=60)
    local_project_directory: Path
    make_command: Sequence[str] = Field(default_factory=lambda: ["make"])
    run_command: Sequence[str] = Field(default_factory=lambda: ["./main"])
    log_file_path: Path
    state_file_path: Path

    @field_validator("local_project_directory", "log_file_path", "state_file_path")
    @classmethod
    def expand_paths(cls, value: Path) -> Path:
        return value.expanduser().resolve()


class RemoteCommandFile(BaseModel):
    filename: str
    url: Optional[AnyHttpUrl] = None
    discovered_at: datetime

    @field_validator("filename")
    @classmethod
    def require_txt_file(cls, value: str) -> str:
        if not value.endswith(".txt"):
            raise ValueError("Command files must end with .txt")
        return value


class ParsedCommand(BaseModel):
    command_id: str
    command_name: str
    scheduled_time: time
    source_filename: str
    discovered_at: datetime

    @field_validator("command_name")
    @classmethod
    def require_snake_case(cls, value: str) -> str:
        if not value.replace("_", "").islower() or "__" in value:
            raise ValueError("Command name must be snake_case")
        return value


class CommandValidationResult(BaseModel):
    command: Optional[ParsedCommand] = None
    decision: CommandDecision
    reason: str


class CommandExecutionRequest(BaseModel):
    command: ParsedCommand
    settings: AgentSettings


class CommandExecutionResult(BaseModel):
    command_id: str
    command_name: str
    status: ExecutionStatus
    decision: CommandDecision
    started_at: datetime
    finished_at: datetime
    message: str = ""
    exit_code: Optional[int] = None


class ICommandSource(Protocol):
    def list_command_files(self) -> Iterable[RemoteCommandFile]:
        """Return command files currently visible from the remote command source."""


class ICommandFilenameParser(Protocol):
    def parse(self, command_file: RemoteCommandFile) -> CommandValidationResult:
        """Parse one remote command filename into a structured command."""


class IFreshnessPolicy(Protocol):
    def evaluate(self, command: ParsedCommand, now: datetime) -> CommandValidationResult:
        """Return EXECUTE only when the command is recent enough to run."""


class ICommandStateStore(Protocol):
    def has_seen(self, command_id: str) -> bool:
        """Return True if this command was already seen or executed."""

    def mark_seen(self, command_id: str) -> None:
        """Record that the command was seen."""

    def mark_executed(self, result: CommandExecutionResult) -> None:
        """Record the final execution result."""


class ICommandHandler(Protocol):
    command_name: str

    def execute(self, request: CommandExecutionRequest) -> CommandExecutionResult:
        """Execute one approved command."""


class ICommandRegistry(Protocol):
    def get_handler(self, command_name: str) -> Optional[ICommandHandler]:
        """Return the handler for an allowlisted command name, or None."""

    def registered_commands(self) -> Sequence[str]:
        """Return the names of all allowlisted commands."""


class ICommandDispatcher(Protocol):
    def dispatch(self, command: ParsedCommand) -> CommandExecutionResult:
        """Route a valid parsed command to the correct handler."""


class IProcessRunner(Protocol):
    def run(self, args: Sequence[str], cwd: Path) -> tuple[int, str, str]:
        """Run a local allowlisted process and return exit_code, stdout, stderr."""


class IAgentLogger(Protocol):
    def info(self, message: str, **context: object) -> None:
        """Write informational audit details."""

    def warning(self, message: str, **context: object) -> None:
        """Write warning audit details."""

    def error(self, message: str, **context: object) -> None:
        """Write error audit details."""

7. Example Handler Shape

The implementation should use small handler classes instead of a giant if/else block.

class MakeRunHandler:
    command_name = "make_run"

    def __init__(self, process_runner: IProcessRunner):
        self.process_runner = process_runner

    def execute(self, request: CommandExecutionRequest) -> CommandExecutionResult:
        started_at = datetime.now()

        make_exit, make_stdout, make_stderr = self.process_runner.run(
            request.settings.make_command,
            cwd=request.settings.local_project_directory,
        )

        if make_exit != 0:
            finished_at = datetime.now()
            return CommandExecutionResult(
                command_id=request.command.command_id,
                command_name=request.command.command_name,
                status=ExecutionStatus.FAILED,
                decision=CommandDecision.FAILED,
                started_at=started_at,
                finished_at=finished_at,
                message=make_stderr or make_stdout,
                exit_code=make_exit,
            )

        run_exit, run_stdout, run_stderr = self.process_runner.run(
            request.settings.run_command,
            cwd=request.settings.local_project_directory,
        )

        finished_at = datetime.now()
        return CommandExecutionResult(
            command_id=request.command.command_id,
            command_name=request.command.command_name,
            status=ExecutionStatus.SUCCEEDED if run_exit == 0 else ExecutionStatus.FAILED,
            decision=CommandDecision.EXECUTE if run_exit == 0 else CommandDecision.FAILED,
            started_at=started_at,
            finished_at=finished_at,
            message=run_stdout if run_exit == 0 else run_stderr,
            exit_code=run_exit,
        )

8. Unit Test Checklist

9. Security and Reliability Guardrails

• Use HTTPS when possible.
• Keep the base URL configurable.
• Use an allowlist command registry.
• Do not pass remote file contents directly to subprocess.
• Log malformed and unknown command files instead of failing silently.
• Store local execution history to prevent duplicate runs.
• Consider adding signed JSON command files in the next version.
• Consider adding a separate status upload/reporting channel later, but do not block this prototype on it.

10. Recommended Milestones

1. Milestone 1: Create project structure, config model, and Pydantic contracts.
2. Milestone 2: Implement filename parser and freshness policy with unit tests.
3. Milestone 3: Implement HTTP command source adapter and local state store.
4. Milestone 4: Implement command registry, dispatcher, and health_check handler.
5. Milestone 5: Implement make_run handler with safe subprocess runner.
6. Milestone 6: Add README, .env.example, installation notes, and systemd service notes.

11. Final Instruction to the Implementation Agent

Build this prototype cleanly and test-first. Keep the design modular. Do not create a remote shell. Treat the website command folder as an untrusted queue of command requests. Only parse, validate, and dispatch known command names through registered handler classes. Put files in the project structure described above unless the existing system already has a better established location. If a different location is chosen, document the reason in README.md.