kingswatching（kingswatching 搜索）

King's Watching - AI Workflow Enforcer + Task Translator

The Three Core Problems We Solve

v0.4.0 Major Upgrade: Task Translator + Progress Reports

The Problem

User says: "Download 100 research reports"

• Human understands: This is one task
• AI understands: "100 is too many, I'll do 14 and call it done"
• Result: Only 14 downloaded, 86 "intelligently skipped"

Solution: TaskTranslator

from overseer import TaskTranslator, translate_and_run

# Natural language command
command = "Download 100 photochemistry research reports"

# Auto translate and execute
result = translate_and_run(command)

Execution Process:

User: "Download 100 reports"
    ↓
TaskTranslator parses intent
    ↓
Identifies: batch download task, target 100 items
    ↓
Calculates: 10 items per batch → need 10 rounds
    ↓
Generates 10 Steps:
  Step 1: Download 1-10 (verify: must complete 10)
  Step 2: Download 11-20 (verify: must complete 10)
  ...
  Step 10: Download 91-100 (verify: must complete 10)
    ↓
King's Watching executes sequentially
    ↓
✅ All 100 completed

Key Innovation: Every Step has forced verification, AI cannot cut corners

Core Usage Patterns

Pattern 1: One-liner Natural Language (Recommended)

from overseer import translate_and_run

# Download 100 reports
result = translate_and_run("Download 100 photochemistry research reports")

# Write 100k word report
result = translate_and_run("Write a 100,000-word industry research report")

# Analyze 1000 data entries
result = translate_and_run("Analyze 1000 user feedback entries")

Pattern 2: Translate First, Then Execute

from overseer import TaskTranslator, Overseer

# Translate natural language
translator = TaskTranslator()
plan = translator.translate("Download 100 reports")

# View execution plan
print(translator.explain_plan(plan))
# 📋 Task Translation Result
# Original command: Download 100 reports
# Task scale: 100 items
# Estimated time: 3000 seconds
# Auto split into 10 execution steps:
#   Step 1: Process 1-10 (of 100)
#          └─ Verify: must complete 10 items
#   Step 2: Process 11-20 (of 100)
#          └─ Verify: must complete 10 items
# ...
# ✅ Each Step has forced verification, AI cannot cut corners

# Create Overseer and execute
workflow = Overseer.from_plan(plan)
result = workflow.run()

Pattern 3: Traditional (Manual Step Definition)

from overseer import Overseer

workflow = Overseer("data_analysis")

@workflow.step("Fetch Data")
def step1(ctx): 
    return download_data()

@workflow.step("Analyze Data", heartbeat_interval=68)
def step2(ctx):
    for i, item in enumerate(items):
        if i % 10 == 0:
            ctx.heartbeat(f"Processed {i}/{len(items)} items...")
        analyze(item)
    return results

@workflow.step("Generate Report")
def step3(ctx): 
    return generate_report()

workflow.run()

AI Capacity Limits Configuration

TaskTranslator has built-in AI capacity limits (prevents overload):

Custom Configuration:

from overseer import TaskTranslator

custom_capacity = {
    "search_download": {
        "max_items": 15,
        "time_limit": 400,
        "verification": "count_check"
    }
}

translator = TaskTranslator(capacity_config=custom_capacity)

Intent Pattern Library

TaskTranslator has built-in common task pattern recognition:

Custom Patterns:

from overseer import TaskTranslator

custom_patterns = [
    {
        "name": "batch_crawl",
        "regexes": [r"Crawl (\d+) web pages", r"Scrape (\d+) data entries"],
        "task_type": "web_crawl",
        "unit": "items",
        "parameters": {"respect_robots": True}
    }
]

translator = TaskTranslator(patterns=custom_patterns)

Step Verification Mechanism

Verification Types

# Count verification (download/analysis tasks)
{
    "type": "count_check",
    "min_required": 10,
    "max_retries": 3,
    "on_failure": "retry_this_chunk"
}

# Word count verification (writing tasks)
{
    "type": "word_count_check",
    "min_required": 2000,
    "max_retries": 3
}

# Completeness verification (data processing tasks)
{
    "type": "completeness_check",
    "required_fields": ["field1", "field2"]
}

Failure Handling

@workflow.step("Batch Download", verification={
    "type": "count_check",
    "min_required": 10,
    "on_failure": "retry",
    "max_retries": 3
})
def download(ctx):
    # If download count < 10, auto retry
    # After 3 retries, mark as failed
    pass

Long-running Task Anti-Timeout

Problem: 15-minute Timeout

User: Analyze 1000 financial reports
Agent: OK, starting analysis...
      [10 minutes later...]
      [15 minutes later... System: Is this agent dead? Kill!]
User: ?

Solution A: Heartbeat Mode (Recommended)

@workflow.step("Batch Analysis", heartbeat_interval=68)  # Report every 68s

def analyze(ctx):
    for i, file in enumerate(files):
        if i % 10 == 0:
            ctx.heartbeat(f"Processed {i}/{len(files)}...")
    return results

Solution B: Background Execution Mode

job = workflow.run_async(
    notify_on_complete=True,
    notify_channel="discord"
)

print(f"Job started: {job.id}")
print(f"ETA: {job.eta}")

Complete Examples

Example 1: Report Collection (Auto-chunking)

from overseer import translate_and_run

# One-liner starts complete workflow
result = translate_and_run(
    "Download 100 photochemistry industry reports covering policy, market, tech, and companies",
    notify_on_complete=True,
    notify_channel="discord"
)

# Output:
# 📋 Task Translation Result
# Original command: Download 100 photochemistry industry reports...
# Task scale: 100 items
# Estimated time: 3000 seconds
# Auto split into 10 execution steps:
#   Step 1: Process 1-10 (of 100)
#   Step 2: Process 11-20 (of 100)
# ...
# ✅ Each Step has forced verification, AI cannot cut corners
#
# ⏳ Step 1/10: download_batch_1...
#    💓 Processed 5/10 items...
#    💓 Processed 10/10 items...
# ✅ Step 1/10 complete
#
# ⏳ Step 2/10: download_batch_2...
# ...
# 🎉 All complete!

Example 2: Report Writing (Auto-chunking)

from overseer import translate_and_run

result = translate_and_run("Write a 100,000-word deep research report on photochemistry industry")

# Auto split into 50 Steps (2000 words each)
# Each Step verifies word count before proceeding

API Reference

TaskTranslator

class TaskTranslator:
    def translate(self, natural_command: str, context: Dict = None) -> Dict:
        """Natural language → YAML execution plan"""
        
    def explain_plan(self, plan: Dict) -> str:
        """Generate human-readable plan description"""

Overseer.from_plan

@classmethod
def from_plan(cls, plan: Dict, **kwargs) -> "Overseer":
    """Create Overseer instance from translation plan"""

translate_and_run

def translate_and_run(natural_command: str, **kwargs) -> Dict:
    """One-liner translate and execute"""

Integration with OpenClaw

# Use OpenClaw cron for chunked scheduling
cron.add(
    schedule={"kind": "every", "everyMs": 600000},  # Check every 10 min
    payload={
        "kind": "agentTurn",
        "message": "Check King's Watching task progress",
        "workflow_id": workflow_id
    }
)

# Use OpenClaw messaging to notify user
message.send(
    channel="discord",
    to=user_id,
    message="🎉 Task complete!",
    file=package_path
)

ProgressReporter (Periodic Reports)

Long-running tasks need progress visibility. King's Watching v0.4.0 adds periodic reporting with natural language interval configuration.

Core Features

Usage

from overseer import Overseer, translate_and_run

# Pattern 1: Default 15-min reports
wf = Overseer("My Workflow")

# Pattern 2: Natural language interval
wf = Overseer(
    "My Workflow",
    report_interval="Every 5 minutes"
)

# Pattern 3: Combined with TaskTranslator
translate_and_run(
    "Download 100 reports",
    report_interval="Every 10 minutes"
)

Supported Interval Formats

# English
"Every 5 minutes"     # → 300 seconds
"Every 10 minutes"    # → 600 seconds
"Quarterly"           # → 900 seconds
"Every 30 seconds"    # → 30 seconds
"Every 1 hour"        # → 3600 seconds

Sample Report Output

============================================================
📊 [14:30:00] Progress Report #3
============================================================
Task: auto_batch_xxx
Overall Progress: 5/10 (50.0%)
Elapsed: 25 minutes
ETA: 25 minutes
Current Step: download_batch_5
Step Status: 5 complete / 10 total
============================================================

Disable Reports

wf = Overseer(
    "Silent Task",
    report_progress=False  # Disable reporting
)

Summary

King's Watching v0.4.0 solves five problems:

1. Forced Sequence → Hard constraints, AI cannot skip steps
2. Timeout Disconnect → Heartbeat + background execution + async notification
3. Continuity Loss → State persistence + checkpoint resume
4. Cutting Corners → TaskTranslator auto-chunking + step verification
5. Progress Black Box → ProgressReporter periodic reports (natural language intervals)

Core Value: Enables AI to reliably execute complex, multi-step, large-workload tasks.

One-liner summary:

User says "Download 100", King's Watching auto-splits into 10 Steps, reports every 5 minutes, AI cannot cut corners.