tax

Original source / Raw SKILL.md

---
name: tax
description: Local-first year-round tax memory system for individuals, freelancers, and small businesses. Use when users mention tax documents, receipts, expenses, tax notices, filing preparation, missing forms, accountant meetings, or year-end organization. Captures tax-relevant facts as they happen, tracks what may be missing, and prepares CPA-ready handoff summaries. NEVER provides tax advice, legal interpretations, filing positions, or final tax calculations.
---

# Tax

A local-first, year-round tax memory system.

Tax is designed to help users capture tax-relevant facts throughout the year, prevent missing documents, and prepare organized handoff packages for CPAs, EAs, accountants, or tax software.

This skill is not a tax advisor.  
It is a tax fact capture, organization, and handoff system.

## Core Product Principle

Tax problems are usually not calculation problems first.  
They are information fragmentation problems first.

By the time filing season arrives, users often already have the necessary information — but it is scattered across email, paper mail, receipts, payment platforms, brokerages, bank accounts, and memory.

This skill exists to:
- capture tax-relevant facts early
- preserve them in structured local records
- track what may be missing
- prepare clean outputs for professional review

## What This Skill Does

This skill can:
- Capture tax-relevant events from natural language
- Log tax documents as they are received
- Track expenses and receipts that may matter during filing
- Record tax authority notices
- Maintain questions for a CPA or tax professional
- Compare current-year records with prior-year patterns
- Surface potentially missing forms or incomplete records
- Generate structured year-end and pre-filing handoff summaries
- Store everything locally

This skill cannot:
- Provide tax advice
- Interpret tax law
- Recommend filing positions
- Determine whether a specific item is deductible
- Calculate final tax liability
- Replace a CPA, EA, attorney, or licensed tax professional

## Safety Boundary: Facts vs. Judgments

This skill records facts.  
It does not make legal or tax judgments.

Examples of facts:
- "Received a 1099-NEC from Client A"
- "Spent $120 on a client meal today"
- "Received an IRS notice"
- "Need to ask CPA about contractor payment treatment"

Examples of judgments this skill does NOT make:
- whether a payment is deductible
- what percentage may be deductible
- whether a filing position is appropriate
- how a tax authority will interpret a record

When users ask judgment questions, this skill should:
1. Record the fact or question
2. Mark it for professional review
3. Encourage confirmation with a licensed professional

## Privacy & Storage

All data is stored locally only.

Base path:  
`~/.openclaw/workspace/memory/tax/`

No cloud storage is required.  
No tax authority systems are accessed.  
No external APIs are required for storage.  
No documents are uploaded unless the user independently chooses to do so outside this skill.

## Tax Memory Model

This skill organizes data into six local layers:

### 1. Event Capture Layer
`ledger_events.json`

Raw tax-relevant facts captured from natural language:
- expenses
- documents received
- notices
- questions for CPA
- reminders
- unknown tax-relevant events

### 2. Document Inventory
`documents.json`

Formal forms and document records:
- W-2
- 1099 series
- K-1
- mortgage interest statements
- property tax statements
- donation receipts
- brokerage tax forms
- other filing-year forms

### 3. Expected Documents
`expected_documents.json`

Predicted or expected forms based on:
- prior-year history
- recurring issuers
- user-declared accounts or entities
- manually added expectations

### 4. Expense & Receipt Records
`expenses.json`

Structured expense or receipt facts that may need professional review later.

### 5. Notices & Questions
`notices.json`  
`questions_for_cpa.json`

Tracks:
- tax authority notices
- unresolved follow-up items
- questions the user wants to ask a CPA

### 6. Year State
`year_state.json`

Tracks annual status:
- capturing
- reconciling
- pre_filing
- filed
- archived
- notice_followup

## Product Behaviors

### Frictionless Capture
Users should be able to speak naturally.

Example:  
"Today I spent 120 dollars taking a client to lunch."

The skill should convert that into structured local records with minimal follow-up, preserving raw text even when some fields remain uncertain.

Capture first.  
Refine later.

### Cross-Year Memory
Prior-year records help predict current-year missing items.

Example:  
"If Robinhood issued a 1099-B last year but none has been logged this year, surface that as a possible missing document."

This is a reminder system, not a legal conclusion.

### CPA Handoff
The final output of the skill is not tax advice.  
It is a clean handoff package for professional review.

A handoff package may include:
- filing snapshot
- income document inventory
- expense summary by category
- outstanding notices
- missing items
- questions for CPA

## Recommended Usage

### During the Year
Use this skill when:
- a form arrives
- a receipt or expense happens
- a tax notice is received
- a tax-related question comes up
- the user wants to avoid losing track of details

### Before Filing
Use this skill to:
- compare expected vs received forms
- surface missing items
- summarize expense categories
- prepare a CPA handoff package
- collect unresolved questions

### After Filing
Use this skill to:
- archive the year
- mark unresolved notices
- carry forward expected recurring documents

## Core Workflows

### 1. Capture a tax-relevant event
Example user messages:
- "Today I paid $49 for Adobe"
- "I received a 1099 from Stripe"
- "IRS sent me a letter"
- "Remind me to ask my CPA about this contractor payment"

Internal action:
- classify event
- save raw text
- extract structured fields where possible
- store locally
- mark uncertain items for follow-up

### 2. Log a document
Use when a formal tax form or relevant supporting document is received.

### 3. Record an expense or receipt
Use when the user mentions a business, rental, freelance, or otherwise tax-relevant payment or receipt.

Important:  
The skill records the fact and category.  
It does not determine tax treatment.

### 4. Record a notice
Use when the user mentions receiving communication from a tax authority.

### 5. Check missing items
Use prior-year memory and current-year records to surface items that may still be missing.

### 6. Prepare CPA handoff package
Generate a structured Markdown handoff summary for professional review.

### 7. Generate annual summary
Generate Markdown and CSV annual summary outputs for review, recordkeeping, or handoff support.
## Files

- `ledger_events.json` — captured raw tax-relevant events
- `documents.json` — formal document inventory
- `expected_documents.json` — expected or predicted forms
- `expenses.json` — structured expense records
- `notices.json` — authority notices
- `questions_for_cpa.json` — open professional review questions
- `year_state.json` — annual workflow state
- `summaries/` — generated Markdown and CSV handoff outputs

## Scripts

| Script | Purpose |
|--------|---------|
| `capture_event.py` | Main entrypoint for tax-relevant natural language capture |
| `add_document.py` | Log a formal tax document |
| `track_expense.py` | Record an expense or receipt fact |
| `log_notice.py` | Record a tax authority notice |
| `add_cpa_question.py` | Save a question for professional review |
| `check_missing.py` | Compare prior-year and current-year document history to surface possible missing documents |
| `prep_meeting.py` | Generate CPA-ready handoff package |
| `generate_summary.py` | Produce Markdown and CSV annual summary outputs |
| `archive_year.py` | Archive a filing year and roll forward expected items |
| `set_year_state.py` | Update annual tax workflow status |

## Response Style Rules

When using this skill:
- Prefer operational clarity over explanation
- Capture facts first, even if incomplete
- Preserve raw user wording when helpful
- Clearly distinguish recorded facts from unresolved judgments
- Use phrases like:
  - "recorded"
  - "captured"
  - "flagged for professional review"
  - "possible missing item"
- Avoid phrases like:
  - "deductible"
  - "allowed deduction"
  - "you should file"
  - "you owe"
  - "safe harbor"
  - "final liability"

## Standard Boundary Response

If a user asks:
- "Can I deduct this?"
- "How much tax do I owe?"
- "Should I file this as X or Y?"
- "Will the IRS accept this?"

The skill should respond in this pattern:
1. Record the underlying fact
2. Offer to log it as a CPA review item
3. Explain that final tax treatment requires a licensed professional

## Disclaimer

This skill is for organization, recordkeeping, and professional handoff only.

Tax outcomes depend on jurisdiction, dates, filing status, entity structure, elections, and documentation quality.  
Always confirm tax treatment, filing positions, and calculations with a licensed CPA, EA, attorney, or other qualified tax professional.


---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "agenticio",
  "slug": "tax",
  "displayName": "Tax",
  "latest": {
    "version": "4.0.0",
    "publishedAt": 1773239727138,
    "commit": "https://github.com/openclaw/skills/commit/ea6c8e4ce620ce282b2735bcb52c95d5eebc2178"
  },
  "history": [
    {
      "version": "2.1.0",
      "publishedAt": 1773065739758,
      "commit": "https://github.com/openclaw/skills/commit/2762cbe2572a06674b56349303ee1087824424df"
    },
    {
      "version": "2.0.0",
      "publishedAt": 1772975474776,
      "commit": "https://github.com/openclaw/skills/commit/55a2646514b753d7fca736f0b46633ec257f8b4a"
    }
  ]
}

```

### references/cpa-handoff.md

```markdown
# CPA Handoff

## Why the Handoff Matters

For many users, the final consumer of tax records is not the user.

It is usually:
- a CPA
- an EA
- an accountant
- a bookkeeper
- a tax attorney
- tax software used during filing

That means the skill's value is not only in what it captures.
Its value is also in how clearly it can hand those records off.

The better the handoff, the more useful the skill becomes.

## Core Product Principle

Messy tax memory should become clean professional review material.

The handoff package should reduce:
- confusion
- missing records
- repeated back-and-forth questions
- last-minute filing stress

It should increase:
- clarity
- completeness
- confidence
- speed of professional review

## What a Handoff Package Should Include

A strong handoff package should contain at least these sections:

### 1. Filing Snapshot
A concise overview of the year:
- tax year
- current state
- documents received count
- expected documents still missing
- open notices
- open CPA questions

### 2. Income Documents Inventory
A list of formal documents received or expected, such as:
- W-2
- 1099 series
- K-1
- brokerage forms
- mortgage interest statements
- other key filing-year documents

Each row should show:
- document type
- issuer
- amount if known
- status

### 3. Expense Summary by Category
A summary of recorded tax-relevant expenses grouped by category.

Examples:
- advertising
- meals
- travel
- software
- office supplies
- contractor payments

Important:
This is an organizational summary, not a deduction summary.

### 4. Notices and Follow-Up Items
A summary of all recorded tax authority notices and unresolved follow-up needs.

### 5. Questions for My CPA
A clean list of unresolved items the user wants professional help with.

## Output Formats

The skill should prefer structured formats such as:
- Markdown for human review
- CSV for import into spreadsheets or other systems

Recommended outputs:
- `summary_<year>.md`
- `summary_<year>.csv`

## Safe Language

Use:
- filing snapshot
- tax-relevant expense summary
- recorded documents
- possible missing items
- questions for CPA
- professional review items

Avoid:
- deductible total
- tax savings
- final tax due
- correct filing position
- allowed deduction amount

## Example Value Moment

The ideal outcome is that a CPA or tax preparer can review the package and say:

"This is clear, organized, and easy to work with."

That moment is one of the strongest signs that the skill is delivering real value.

## Product Standard

A handoff package should be:
- easy to scan
- professionally structured
- honest about missing information
- clear about unresolved issues
- safe in its wording
- ready for professional review without pretending to replace it

```

### references/cross-year-memory.md

```markdown
# Cross-Year Memory

## Why Cross-Year Memory Matters

Tax recordkeeping is not a one-time activity.
It is a repeating annual cycle.

What happened last year often predicts what should be expected this year:
- recurring forms
- recurring issuers
- recurring brokerages
- recurring vendors
- recurring questions for a CPA
- recurring documentation gaps

A useful tax skill should not only store this year's records.
It should also help the user notice what may still be missing.

## Core Principle

Prior-year behavior is not proof.
It is a signal.

This skill uses prior-year patterns to surface possible missing items, not to make legal conclusions.

Examples:
- A brokerage issued a 1099-B last year, but none has been logged yet this year
- A bank issued a 1099-INT last year, but this year no corresponding document has been recorded
- The user logged recurring software expenses last year, but key months are missing this year
- The user usually asks a CPA about contractor treatment, but no question has been logged yet for the current year

These are reminder signals only.

## Expected Documents

The skill should maintain an `expected_documents.json` layer that tracks forms or records likely to appear in the current year.

Sources for expectations may include:
- prior-year document history
- recurring issuers
- user-declared accounts
- manually added expected forms
- previous accountant handoff packages

Expected items should be marked with statuses such as:
- awaiting
- received
- no_longer_expected
- needs_review

## Safe Use of Cross-Year Memory

Cross-year memory should be used to:
- flag possible missing forms
- prompt the user to confirm whether an issuer is still relevant
- improve filing readiness
- reduce forgotten documents
- reduce repeated annual chaos

Cross-year memory should not be used to:
- assume a legal obligation
- state that a form must exist
- interpret why a document is missing
- conclude tax treatment

## Example User-Facing Language

Preferred:
- "You received a Robinhood 1099-B last year, but none has been logged this year yet."
- "This may be a missing document worth checking."
- "Would you like me to keep this on your expected documents list?"

Avoid:
- "You are required to file this form"
- "You definitely should have received this already"
- "This means your taxes are incomplete"
- "The IRS expects this from you"

## Product Value

Cross-year memory creates trust because it helps the user answer one of the most painful tax questions:

"What am I forgetting?"

The skill becomes much more valuable when it can surface likely omissions before filing season becomes urgent.

## Design Rule

Treat prior-year history as predictive memory, not authority.

The skill should help the user remember patterns across years while leaving legal interpretation and filing conclusions to licensed professionals.

```

### references/fact-vs-judgment.md

```markdown
# Facts vs. Judgments

## Why This Boundary Exists

Tax treatment depends on jurisdiction, timing, filing status, entity structure, elections, documentation quality, and professional interpretation.

Because of that, this skill must separate:

- facts that can be recorded safely
- judgments that require licensed professional review

This boundary protects:
- the user
- the skill publisher
- the platform
- the long-term reliability of the product

## Facts This Skill Can Record

Examples of safe fact capture:
- a payment happened
- a receipt was received
- a tax document arrived
- a tax authority sent a notice
- the user wants to ask a CPA a question
- a document appears to be missing compared with a prior year

Examples:
- "Received a 1099-NEC from Client A"
- "Paid $49 for Adobe on March 1"
- "Got an IRS letter today"
- "Need to ask my CPA about contractor payments"

These are recordkeeping facts.
They are appropriate for local storage and later review.

## Judgments This Skill Must Not Make

This skill must not decide:
- whether an expense is deductible
- how much of an item may be deductible
- whether a filing position is correct
- whether a user owes tax
- what a tax authority will accept
- whether a user should choose one treatment over another

Examples of prohibited behavior:
- "This is definitely deductible"
- "You can deduct 50% of this meal"
- "You should file this as business income"
- "Your estimated tax this quarter should be $X"
- "The IRS will probably accept this"

## Safe Pattern for User Questions

When a user asks a judgment question such as:
- "Can I deduct this?"
- "How should I file this?"
- "Will this count as a business expense?"
- "How much do I owe?"

The skill should respond by:
1. recording the underlying fact
2. offering to log it for CPA review
3. stating clearly that final tax treatment requires a licensed professional

## Preferred Language

Use:
- recorded
- captured
- logged
- flagged for professional review
- possible missing item
- tax-relevant record
- question for CPA

Avoid:
- deductible
- guaranteed deduction
- should file
- you owe
- allowed by IRS
- safe harbor
- final liability

## Product Principle

The skill is most valuable when it becomes the user's trusted fact archive.

Facts are durable.
Judgments are conditional.

The skill should preserve durable facts and hand conditional judgments to professionals.

```

### references/recordkeeping.md

```markdown
# Recordkeeping

## Why Recordkeeping Comes First

Most tax stress does not begin at filing.
It begins when records are lost, delayed, scattered, or forgotten.

By filing season, many users already had the necessary information at some point:
- a receipt
- a payment confirmation
- a brokerage form
- a donation acknowledgment
- a tax notice
- a question they meant to ask a CPA

The problem is not always lack of information.
The problem is often lack of timely capture and organized retention.

## Core Principle

Capture early.
Store locally.
Refine later.

A good tax recordkeeping system should prioritize preserving facts at the moment they appear, even if some fields are incomplete.

## What Should Be Recorded

This skill is designed to capture and organize tax-relevant records such as:

### Documents
- W-2
- 1099 forms
- K-1 forms
- mortgage interest statements
- brokerage tax forms
- donation acknowledgments
- property tax statements

### Expenses and Receipts
- software subscriptions
- meals
- travel
- advertising
- office supplies
- contractor-related payments
- other tax-relevant business or personal records the user wants archived

### Notices
- IRS letters
- state tax notices
- local tax correspondence
- unresolved authority follow-ups

### Questions
- questions for CPA review
- unresolved treatment questions
- filing-season follow-up items

## Minimum Useful Fields

When possible, records should capture fields such as:
- date
- amount
- issuer or merchant
- category
- raw user wording
- tax year
- notes
- follow-up status

If some fields are missing, the skill should still preserve the event and mark it for later review.

## Local-First Storage

All records should remain local to the user's workspace unless the user independently chooses another workflow outside this skill.

Recommended base path:
`~/.openclaw/workspace/memory/tax/`

This helps preserve privacy and supports long-term control over retention.

## Incomplete Records Are Better Than Lost Records

An incomplete record can still be reviewed later.
A lost record usually cannot.

That is why the product should prefer:
- fast capture
- raw text preservation
- confidence tracking
- follow-up flags

over:
- forcing every field to be perfect before saving

## Product Standard

The recordkeeping layer should help the user:
- avoid forgetting
- avoid re-creating facts from memory
- reduce filing-season chaos
- prepare cleaner professional handoff packages

This skill is strongest when it becomes the user's reliable archive of tax-relevant facts across the whole year.

```

### scripts/add_cpa_question.py

```python
#!/usr/bin/env python3
"""Record a question for CPA or tax professional review."""

import argparse
import json
import os
import uuid
from datetime import datetime
from typing import Any, Dict, Optional


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
QUESTIONS_FILE = os.path.join(BASE_DIR, "questions_for_cpa.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_questions() -> Dict[str, Any]:
    if os.path.exists(QUESTIONS_FILE):
        with open(QUESTIONS_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"questions": []}


def save_questions(data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(QUESTIONS_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def infer_tax_year(explicit_tax_year: Optional[int]) -> int:
    if explicit_tax_year is not None:
        return explicit_tax_year
    return datetime.now().year


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Record a question for CPA or tax professional review."
    )
    parser.add_argument(
        "--question",
        required=True,
        help="Question to ask the CPA or tax professional",
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        default=None,
        help="Related tax year",
    )
    parser.add_argument(
        "--linked-record-id",
        default="",
        help="Optional linked document, expense, or notice record ID",
    )
    parser.add_argument(
        "--status",
        default="open",
        choices=["open", "asked", "resolved"],
        help="Current status of the question",
    )
    parser.add_argument(
        "--notes",
        default="",
        help="Optional notes",
    )

    args = parser.parse_args()

    tax_year = infer_tax_year(args.tax_year)
    now_iso = datetime.now().isoformat(timespec="seconds")
    question_id = f"Q-{str(uuid.uuid4())[:8].upper()}"

    linked_record_ids = []
    if args.linked_record_id.strip():
        linked_record_ids.append(args.linked_record_id.strip())

    question = {
        "id": question_id,
        "tax_year": tax_year,
        "question": args.question,
        "linked_record_ids": linked_record_ids,
        "status": args.status,
        "notes": args.notes,
        "created_at": now_iso,
    }

    data = load_questions()
    data["questions"].append(question)
    save_questions(data)

    print(f"Recorded CPA question: {question_id}")
    print(f"  Tax year: {tax_year}")
    print(f"  Status: {args.status}")
    if linked_record_ids:
        print(f"  Linked record IDs: {', '.join(linked_record_ids)}")
    print(f"  Saved to: {QUESTIONS_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/add_document.py

```python
#!/usr/bin/env python3
"""Log a formal tax document into local tax memory."""

import argparse
import json
import os
import uuid
from datetime import datetime
from typing import Any, Dict, Optional

BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_documents() -> Dict[str, Any]:
    if os.path.exists(DOCUMENTS_FILE):
        with open(DOCUMENTS_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"documents": []}


def save_documents(data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(DOCUMENTS_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

def infer_tax_year(date_received: Optional[str], explicit_tax_year: Optional[int]) -> int:
    if explicit_tax_year is not None:
        return explicit_tax_year
    if date_received:
        return datetime.strptime(date_received, "%Y-%m-%d").year
    return datetime.now().year


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Log a formal tax document into local tax memory."
    )
    parser.add_argument(
        "--document-type",
        required=True,
        help="Document type, e.g. W-2, 1099-NEC, 1099-B, K-1",
    )
    parser.add_argument(
        "--issuer",
        required=True,
        help="Issuer of the document, e.g. employer, client, brokerage",
    )
    parser.add_argument(
        "--amount",
        type=float,
        default=None,
        help="Amount shown on the document, if known",
    )
    parser.add_argument(
        "--date-received",
        default=datetime.now().strftime("%Y-%m-%d"),
        help="Date the document was received in YYYY-MM-DD format",
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        default=None,
        help="Tax year the document belongs to",
    )
    parser.add_argument(
        "--status",
        default="received",
        choices=["received", "awaiting", "needs_review", "archived"],
        help="Current status of the document record",
    )
    parser.add_argument(
        "--source-channel",
        default="unknown",
        help="Where the document was received from, e.g. mail, email, portal",
    )
    parser.add_argument(
        "--notes",
        default="",
        help="Optional notes",
    )
    parser.add_argument(
        "--expected",
        action="store_true",
        help="Mark this document as expected or recurring",
    )

    args = parser.parse_args()

    tax_year = infer_tax_year(args.date_received, args.tax_year)
    now_iso = datetime.now().isoformat(timespec="seconds")
    doc_id = f"DOC-{str(uuid.uuid4())[:8].upper()}"

    document = {
        "id": doc_id,
        "tax_year": tax_year,
        "document_type": args.document_type,
        "issuer": args.issuer,
        "amount": args.amount,
        "date_received": args.date_received,
        "status": args.status,
        "expected": args.expected,
        "source_channel": args.source_channel,
        "notes": args.notes,
        "created_at": now_iso,
    }

    data = load_documents()
    data["documents"].append(document)
    save_documents(data)

    print(f"Recorded document: {doc_id}")
    print(f"  Tax year: {tax_year}")
    print(f"  Type: {args.document_type}")
    print(f"  Issuer: {args.issuer}")
    if args.amount is not None:
        print(f"  Amount: ${args.amount:,.2f}")
    print(f"  Date received: {args.date_received}")
    print(f"  Status: {args.status}")
    print(f"  Expected: {'yes' if args.expected else 'no'}")
    print(f"  Saved to: {DOCUMENTS_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/archive_year.py

```python
#!/usr/bin/env python3
"""Archive a tax year and roll forward expected document patterns."""

import argparse
import json
import os
from datetime import datetime
from typing import Any, Dict, List, Tuple


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")
YEAR_STATE_FILE = os.path.join(BASE_DIR, "year_state.json")
EXPECTED_DOCUMENTS_FILE = os.path.join(BASE_DIR, "expected_documents.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def save_json(path: str, data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def filter_by_tax_year(items: List[Dict[str, Any]], tax_year: int) -> List[Dict[str, Any]]:
    return [item for item in items if item.get("tax_year") == tax_year]


def normalized_doc_key(document: Dict[str, Any]) -> Tuple[str, str]:
    document_type = str(document.get("document_type", "")).strip()
    issuer = str(document.get("issuer", "")).strip()
    return (document_type, issuer)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Archive a tax year and roll forward expected document patterns."
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        required=True,
        help="Tax year to archive",
    )

    args = parser.parse_args()
    tax_year = args.tax_year
    next_year = tax_year + 1
    now_iso = datetime.now().isoformat(timespec="seconds")

    documents_data = load_json(DOCUMENTS_FILE, {"documents": []})
    year_state_data = load_json(YEAR_STATE_FILE, {"years": {}})
    expected_data = load_json(EXPECTED_DOCUMENTS_FILE, {"years": {}})

    archived_year_docs = filter_by_tax_year(documents_data.get("documents", []), tax_year)

    # 1. Update current tax year state to archived
    if "years" not in year_state_data:
        year_state_data["years"] = {}

    existing_year_state = year_state_data["years"].get(str(tax_year), {})
    existing_year_state["state"] = "archived"
    existing_year_state["updated_at"] = now_iso
    year_state_data["years"][str(tax_year)] = existing_year_state

    # 2. Prepare expected documents container for next year
    if "years" not in expected_data:
        expected_data["years"] = {}

    next_year_bucket = expected_data["years"].get(str(next_year), {})
    expected_documents = next_year_bucket.get("expected_documents", [])

    existing_keys = {
        (
            str(item.get("document_type", "")).strip(),
            str(item.get("issuer", "")).strip(),
        )
        for item in expected_documents
    }

    added_count = 0

    for doc in archived_year_docs:
        doc_key = normalized_doc_key(doc)
        if doc_key == ("", ""):
            continue
        if doc_key in existing_keys:
            continue

        expected_documents.append(
            {
                "document_type": doc_key[0],
                "issuer": doc_key[1],
                "reason": f"appeared_in_prior_year_{tax_year}",
                "status": "awaiting",
                "source_tax_year": tax_year,
                "created_at": now_iso,
            }
        )
        existing_keys.add(doc_key)
        added_count += 1

    next_year_bucket["expected_documents"] = expected_documents
    next_year_bucket["updated_at"] = now_iso
    expected_data["years"][str(next_year)] = next_year_bucket

    save_json(YEAR_STATE_FILE, year_state_data)
    save_json(EXPECTED_DOCUMENTS_FILE, expected_data)

    print(f"Archived tax year: {tax_year}")
    print(f"  State updated to: archived")
    print(f"  Documents reviewed: {len(archived_year_docs)}")
    print(f"  Expected documents added for {next_year}: {added_count}")
    print(f"  Saved year state to: {YEAR_STATE_FILE}")
    print(f"  Saved expected documents to: {EXPECTED_DOCUMENTS_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/capture_event.py

```python
#!/usr/bin/env python3
"""Capture a tax-relevant natural language event and route it into local tax memory."""

import argparse
import json
import os
import re
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
LEDGER_FILE = os.path.join(BASE_DIR, "ledger_events.json")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")
EXPENSES_FILE = os.path.join(BASE_DIR, "expenses.json")
NOTICES_FILE = os.path.join(BASE_DIR, "notices.json")
QUESTIONS_FILE = os.path.join(BASE_DIR, "questions_for_cpa.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def save_json(path: str, data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def infer_tax_year(explicit_tax_year: Optional[int], date_str: str) -> int:
    if explicit_tax_year is not None:
        return explicit_tax_year
    return datetime.strptime(date_str, "%Y-%m-%d").year


def extract_amount(text: str) -> Optional[float]:
    patterns = [
        r"\$([0-9]+(?:\.[0-9]{1,2})?)",
        r"([0-9]+(?:\.[0-9]{1,2})?)\s*(?:dollars|usd|刀)",
    ]
    for pattern in patterns:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            try:
                return float(match.group(1))
            except ValueError:
                return None
    return None


def classify_event(text: str, extracted_amount: Optional[float]) -> str:
    lower = text.lower()

    if extracted_amount is not None:
        return "expense"

    document_keywords = [
        "1099", "w-2", "w2", "k-1", "k1", "tax form", "tax document"
    ]
    if any(keyword in lower for keyword in document_keywords):
        return "document"

    notice_keywords = [
        "irs", "notice", "letter", "state tax", "tax authority", "penalty"
    ]
    if any(keyword in lower for keyword in notice_keywords):
        return "notice"

    question_keywords = [
        "cpa", "accountant", "ask my cpa", "question", "remind me to ask"
    ]
    if any(keyword in lower for keyword in question_keywords):
        return "question"

    return "unknown"


def confidence_for_event(event_type: str) -> float:
    if event_type == "expense":
        return 0.88
    if event_type in ("document", "notice", "question"):
        return 0.78
    return 0.50


def guess_document_type(text: str) -> str:
    upper = text.upper()
    if "1099-NEC" in upper:
        return "1099-NEC"
    if "1099-B" in upper:
        return "1099-B"
    if "1099-K" in upper:
        return "1099-K"
    if "1099" in upper:
        return "1099"
    if "W-2" in upper or "W2" in upper:
        return "W-2"
    if "K-1" in upper or "K1" in upper:
        return "K-1"
    return "unknown_document"


def guess_issuer(text: str) -> str:
    match = re.search(r"\bfrom\s+([A-Za-z0-9&.\- ]+)", text, re.IGNORECASE)
    if not match:
        return "Unknown"

    issuer = match.group(1).strip().rstrip(".")

    trailing_phrases = [
        " today",
        " yesterday",
        " this morning",
        " this afternoon",
        " tonight",
    ]

    lower_issuer = issuer.lower()
    for phrase in trailing_phrases:
        if lower_issuer.endswith(phrase):
            issuer = issuer[: -len(phrase)].strip()
            break

    return issuer or "Unknown"

def guess_notice_authority(text: str) -> str:
    lower = text.lower()
    if "irs" in lower:
        return "IRS"
    if "state" in lower:
        return "State Tax Authority"
    return "Unknown Authority"


def guess_expense_category(text: str) -> str:
    lower = text.lower()
    if any(word in lower for word in ["lunch", "dinner", "meal", "restaurant", "client to lunch"]):
        return "business_meal"
    if any(word in lower for word in ["adobe", "software", "subscription", "saas"]):
        return "software"
    if "travel" in lower:
        return "travel"
    if "ad" in lower or "advertising" in lower:
        return "advertising"
    return "uncategorized"


def create_expense_record(
    text: str,
    date_str: str,
    tax_year: int,
    amount: Optional[float],
) -> Optional[str]:
    if amount is None:
        return None

    data = load_json(EXPENSES_FILE, {"expenses": []})
    expense_id = f"EXP-{str(uuid.uuid4())[:8].upper()}"
    now_iso = datetime.now().isoformat(timespec="seconds")

    expense = {
        "id": expense_id,
        "tax_year": tax_year,
        "date": date_str,
        "amount": amount,
        "currency": "USD",
        "category": guess_expense_category(text),
        "merchant": "Unknown",
        "purpose": "",
        "documentation_status": "receipt_not_confirmed",
        "professional_review_status": "unreviewed",
        "raw_text": text,
        "created_at": now_iso,
    }

    data["expenses"].append(expense)
    save_json(EXPENSES_FILE, data)
    return expense_id


def create_document_record(
    text: str,
    date_str: str,
    tax_year: int,
    amount: Optional[float],
) -> str:
    data = load_json(DOCUMENTS_FILE, {"documents": []})
    doc_id = f"DOC-{str(uuid.uuid4())[:8].upper()}"
    now_iso = datetime.now().isoformat(timespec="seconds")

    document = {
        "id": doc_id,
        "tax_year": tax_year,
        "document_type": guess_document_type(text),
        "issuer": guess_issuer(text),
        "amount": amount,
        "date_received": date_str,
        "status": "received",
        "expected": False,
        "source_channel": "captured_event",
        "notes": text,
        "created_at": now_iso,
    }

    data["documents"].append(document)
    save_json(DOCUMENTS_FILE, data)
    return doc_id


def create_notice_record(
    text: str,
    date_str: str,
    tax_year: int,
) -> str:
    data = load_json(NOTICES_FILE, {"notices": []})
    notice_id = f"NOT-{str(uuid.uuid4())[:8].upper()}"
    now_iso = datetime.now().isoformat(timespec="seconds")

    notice = {
        "id": notice_id,
        "tax_year": tax_year,
        "authority": guess_notice_authority(text),
        "notice_type": "letter",
        "date_received": date_str,
        "summary": text,
        "response_deadline": None,
        "status": "open",
        "professional_review_needed": True,
        "created_at": now_iso,
    }

    data["notices"].append(notice)
    save_json(NOTICES_FILE, data)
    return notice_id


def create_question_record(
    text: str,
    tax_year: int,
) -> str:
    data = load_json(QUESTIONS_FILE, {"questions": []})
    question_id = f"Q-{str(uuid.uuid4())[:8].upper()}"
    now_iso = datetime.now().isoformat(timespec="seconds")

    question = {
        "id": question_id,
        "tax_year": tax_year,
        "question": text,
        "linked_record_ids": [],
        "status": "open",
        "notes": "Created from capture_event.py",
        "created_at": now_iso,
    }

    data["questions"].append(question)
    save_json(QUESTIONS_FILE, data)
    return question_id


def route_event(
    event_type: str,
    text: str,
    date_str: str,
    tax_year: int,
    amount: Optional[float],
) -> List[str]:
    linked_ids: List[str] = []

    if event_type == "expense":
        expense_id = create_expense_record(text, date_str, tax_year, amount)
        if expense_id:
            linked_ids.append(expense_id)

    elif event_type == "document":
        linked_ids.append(create_document_record(text, date_str, tax_year, amount))

    elif event_type == "notice":
        linked_ids.append(create_notice_record(text, date_str, tax_year))

    elif event_type == "question":
        linked_ids.append(create_question_record(text, tax_year))

    return linked_ids


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Capture a tax-relevant natural language event."
    )
    parser.add_argument(
        "--text",
        required=True,
        help="Natural language tax-relevant text to capture",
    )
    parser.add_argument(
        "--date",
        default=datetime.now().strftime("%Y-%m-%d"),
        help="Event date in YYYY-MM-DD format",
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        default=None,
        help="Related tax year",
    )
    parser.add_argument(
        "--source",
        default="chat",
        help="Source of the event, e.g. chat, note, imported_text",
    )

    args = parser.parse_args()

    extracted_amount = extract_amount(args.text)
    event_type = classify_event(args.text, extracted_amount)
    confidence = confidence_for_event(event_type)
    tax_year = infer_tax_year(args.tax_year, args.date)
    now_iso = datetime.now().isoformat(timespec="seconds")
    event_id = f"EVT-{str(uuid.uuid4())[:8].upper()}"

    linked_record_ids = route_event(
        event_type=event_type,
        text=args.text,
        date_str=args.date,
        tax_year=tax_year,
        amount=extracted_amount,
    )

    event = {
        "id": event_id,
        "tax_year": tax_year,
        "event_type": event_type,
        "raw_text": args.text,
        "date": args.date,
        "amount": extracted_amount,
        "currency": "USD" if extracted_amount is not None else None,
        "source": args.source,
        "confidence": confidence,
        "status": "captured",
        "needs_followup": event_type == "unknown",
        "linked_record_ids": linked_record_ids,
        "created_at": now_iso,
    }

    data = load_json(LEDGER_FILE, {"events": []})
    data["events"].append(event)
    save_json(LEDGER_FILE, data)

    print(f"Captured event: {event_id}")
    print(f"  Tax year: {tax_year}")
    print(f"  Event type: {event_type}")
    print(f"  Date: {args.date}")
    if extracted_amount is not None:
        print(f"  Amount detected: USD {extracted_amount:,.2f}")
    print(f"  Confidence: {confidence:.2f}")
    print(f"  Linked record IDs: {', '.join(linked_record_ids) if linked_record_ids else 'none'}")
    print(f"  Needs follow-up: {'yes' if event['needs_followup'] else 'no'}")
    print(f"  Saved to: {LEDGER_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/check_missing.py

```python
#!/usr/bin/env python3
"""Check for possibly missing tax documents by comparing tax years."""

import argparse
import json
import os
from typing import Any, Dict, List, Set, Tuple


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def normalize_doc_key(document: Dict[str, Any]) -> Tuple[str, str]:
    document_type = str(document.get("document_type", "")).strip()
    issuer = str(document.get("issuer", "")).strip()
    return (document_type, issuer)


def documents_for_year(documents: List[Dict[str, Any]], tax_year: int) -> List[Dict[str, Any]]:
    return [doc for doc in documents if doc.get("tax_year") == tax_year]


def key_set(documents: List[Dict[str, Any]]) -> Set[Tuple[str, str]]:
    return {normalize_doc_key(doc) for doc in documents if normalize_doc_key(doc) != ("", "")}


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Check for possibly missing documents by comparing tax years."
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        required=True,
        help="Current tax year to inspect",
    )
    parser.add_argument(
        "--compare-year",
        type=int,
        required=True,
        help="Prior tax year to compare against",
    )

    args = parser.parse_args()

    data = load_json(DOCUMENTS_FILE, {"documents": []})
    all_documents = data.get("documents", [])

    current_docs = documents_for_year(all_documents, args.tax_year)
    prior_docs = documents_for_year(all_documents, args.compare_year)

    current_keys = key_set(current_docs)
    prior_keys = key_set(prior_docs)

    missing_keys = sorted(prior_keys - current_keys)

    print(f"Tax year checked: {args.tax_year}")
    print(f"Compared against: {args.compare_year}")
    print(f"Documents in {args.compare_year}: {len(prior_docs)}")
    print(f"Documents in {args.tax_year}: {len(current_docs)}")
    print("")

    if not prior_docs:
        print("No prior-year documents found. Nothing to compare.")
        return

    if not missing_keys:
        print("No possible missing documents found based on prior-year document history.")
        return

    print("Possible missing documents:")
    for document_type, issuer in missing_keys:
        print(f"- {document_type} from {issuer}")


if __name__ == "__main__":
    main()

```

### scripts/generate_summary.py

```python
#!/usr/bin/env python3
"""Generate annual tax summaries in Markdown and CSV."""

import argparse
import csv
import json
import os
from datetime import datetime
from typing import Any, Dict, List


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")
EXPENSES_FILE = os.path.join(BASE_DIR, "expenses.json")
NOTICES_FILE = os.path.join(BASE_DIR, "notices.json")
QUESTIONS_FILE = os.path.join(BASE_DIR, "questions_for_cpa.json")
YEAR_STATE_FILE = os.path.join(BASE_DIR, "year_state.json")
SUMMARIES_DIR = os.path.join(BASE_DIR, "summaries")


def ensure_dirs() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)
    os.makedirs(SUMMARIES_DIR, exist_ok=True)


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def filter_by_tax_year(items: List[Dict[str, Any]], tax_year: int) -> List[Dict[str, Any]]:
    return [item for item in items if item.get("tax_year") == tax_year]


def format_money(amount: Any, currency: str = "USD") -> str:
    if amount is None:
        return "Unknown"
    try:
        return f"{currency} {float(amount):,.2f}"
    except (ValueError, TypeError):
        return str(amount)


def summarize_expenses(expenses: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    grouped: Dict[str, Dict[str, Any]] = {}

    for expense in expenses:
        category = expense.get("category", "uncategorized")
        amount = float(expense.get("amount", 0.0))
        currency = expense.get("currency", "USD")

        if category not in grouped:
            grouped[category] = {
                "category": category,
                "count": 0,
                "total": 0.0,
                "currency": currency,
            }

        grouped[category]["count"] += 1
        grouped[category]["total"] += amount

    return [grouped[key] for key in sorted(grouped.keys())]


def count_open_notices(notices: List[Dict[str, Any]]) -> int:
    return len([n for n in notices if n.get("status") != "closed"])


def count_open_questions(questions: List[Dict[str, Any]]) -> int:
    return len([q for q in questions if q.get("status") != "resolved"])


def build_markdown(
    tax_year: int,
    documents: List[Dict[str, Any]],
    expenses: List[Dict[str, Any]],
    notices: List[Dict[str, Any]],
    questions: List[Dict[str, Any]],
    year_state: Dict[str, Any],
) -> str:
    generated_at = datetime.now().isoformat(timespec="seconds")
    expense_summary = summarize_expenses(expenses)

    lines: List[str] = []
    lines.append(f"# Tax Year {tax_year} Annual Summary")
    lines.append("")
    lines.append(f"Generated at: {generated_at}")
    lines.append("")

    lines.append("## Year Overview")
    lines.append("")
    lines.append(f"- Tax year: {tax_year}")
    lines.append(f"- Workflow state: {year_state.get('state', 'unknown')}")
    lines.append(f"- Documents recorded: {len(documents)}")
    lines.append(f"- Expense records: {len(expenses)}")
    lines.append(f"- Open notices: {count_open_notices(notices)}")
    lines.append(f"- Open CPA questions: {count_open_questions(questions)}")
    lines.append("")

    lines.append("## Documents")
    lines.append("")
    if documents:
        lines.append("| Type | Issuer | Amount | Date Received | Status |")
        lines.append("|------|--------|--------|---------------|--------|")
        for doc in documents:
            lines.append(
                f"| {doc.get('document_type', '')} | "
                f"{doc.get('issuer', '')} | "
                f"{format_money(doc.get('amount'))} | "
                f"{doc.get('date_received', '')} | "
                f"{doc.get('status', '')} |"
            )
    else:
        lines.append("_No documents recorded for this tax year._")
    lines.append("")

    lines.append("## Expense Summary by Category")
    lines.append("")
    if expense_summary:
        lines.append("| Category | Count | Total |")
        lines.append("|----------|-------|-------|")
        for row in expense_summary:
            lines.append(
                f"| {row['category']} | {row['count']} | "
                f"{format_money(row['total'], row['currency'])} |"
            )
    else:
        lines.append("_No expenses recorded for this tax year._")
    lines.append("")

    lines.append("## Notices")
    lines.append("")
    if notices:
        lines.append("| Authority | Type | Date Received | Status | Summary |")
        lines.append("|-----------|------|---------------|--------|---------|")
        for notice in notices:
            lines.append(
                f"| {notice.get('authority', '')} | "
                f"{notice.get('notice_type', '')} | "
                f"{notice.get('date_received', '')} | "
                f"{notice.get('status', '')} | "
                f"{notice.get('summary', '')} |"
            )
    else:
        lines.append("_No notices recorded for this tax year._")
    lines.append("")

    lines.append("## Questions for CPA")
    lines.append("")
    if questions:
        for idx, question in enumerate(questions, start=1):
            lines.append(f"{idx}. {question.get('question', '')} ({question.get('status', 'unknown')})")
    else:
        lines.append("_No CPA questions recorded for this tax year._")
    lines.append("")

    lines.append("## Notes")
    lines.append("")
    lines.append(
        "This summary is for organization, recordkeeping, and professional handoff support only. "
        "It does not provide tax advice or legal interpretations."
    )
    lines.append("")

    return "\n".join(lines)

def write_csv(
    output_path: str,
    tax_year: int,
    documents: List[Dict[str, Any]],
    expenses: List[Dict[str, Any]],
    expense_summary: List[Dict[str, Any]],
    notices: List[Dict[str, Any]],
    questions: List[Dict[str, Any]],
    year_state: Dict[str, Any],
) -> None:
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)

        writer.writerow(["section", "field_1", "field_2", "field_3", "field_4", "field_5"])

        writer.writerow(["overview", "tax_year", tax_year, "", "", ""])
        writer.writerow(["overview", "state", year_state.get("state", "unknown"), "", "", ""])
        writer.writerow(["overview", "documents_recorded", len(documents), "", "", ""])
        writer.writerow(["overview", "expense_records", len(expenses), "", "", ""])
        writer.writerow(["overview", "open_notices", count_open_notices(notices), "", "", ""])
        writer.writerow(["overview", "open_questions_for_cpa", count_open_questions(questions), "", "", ""])

        for doc in documents:
            writer.writerow([
                "document",
                doc.get("document_type", ""),
                doc.get("issuer", ""),
                doc.get("amount", ""),
                doc.get("date_received", ""),
                doc.get("status", ""),
            ])

        for row in expense_summary:
            writer.writerow([
                "expense_summary",
                row.get("category", ""),
                row.get("count", 0),
                row.get("total", 0.0),
                row.get("currency", "USD"),
                "",
            ])

        for notice in notices:
            writer.writerow([
                "notice",
                notice.get("authority", ""),
                notice.get("notice_type", ""),
                notice.get("date_received", ""),
                notice.get("status", ""),
                notice.get("summary", ""),
            ])

        for question in questions:
            writer.writerow([
                "question",
                question.get("question", ""),
                question.get("status", ""),
                "",
                "",
                "",
            ])


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate annual tax summaries in Markdown and CSV."
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        required=True,
        help="Tax year to summarize",
    )

    args = parser.parse_args()
    tax_year = args.tax_year

    ensure_dirs()

    documents_data = load_json(DOCUMENTS_FILE, {"documents": []})
    expenses_data = load_json(EXPENSES_FILE, {"expenses": []})
    notices_data = load_json(NOTICES_FILE, {"notices": []})
    questions_data = load_json(QUESTIONS_FILE, {"questions": []})
    year_state_data = load_json(YEAR_STATE_FILE, {"years": {}})

    documents = filter_by_tax_year(documents_data.get("documents", []), tax_year)
    expenses = filter_by_tax_year(expenses_data.get("expenses", []), tax_year)
    notices = filter_by_tax_year(notices_data.get("notices", []), tax_year)
    questions = filter_by_tax_year(questions_data.get("questions", []), tax_year)
    year_state = year_state_data.get("years", {}).get(str(tax_year), {})

    markdown = build_markdown(tax_year, documents, expenses, notices, questions, year_state)
    expense_summary = summarize_expenses(expenses)

    md_path = os.path.join(SUMMARIES_DIR, f"annual_summary_{tax_year}.md")
    csv_path = os.path.join(SUMMARIES_DIR, f"annual_summary_{tax_year}.csv")

    with open(md_path, "w", encoding="utf-8") as f:
        f.write(markdown)

    write_csv(csv_path, tax_year, documents, expenses, expense_summary, notices, questions, year_state)
    print(f"Generated annual summary for tax year {tax_year}")
    print(f"  Documents included: {len(documents)}")
    print(f"  Expenses included: {len(expenses)}")
    print(f"  Notices included: {len(notices)}")
    print(f"  Questions included: {len(questions)}")
    print(f"  Markdown saved to: {md_path}")
    print(f"  CSV saved to: {csv_path}")


if __name__ == "__main__":
    main()

```

### scripts/log_notice.py

```python
#!/usr/bin/env python3
"""Record a tax authority notice into local tax memory."""

import argparse
import json
import os
import uuid
from datetime import datetime
from typing import Any, Dict, Optional


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
NOTICES_FILE = os.path.join(BASE_DIR, "notices.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_notices() -> Dict[str, Any]:
    if os.path.exists(NOTICES_FILE):
        with open(NOTICES_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"notices": []}


def save_notices(data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(NOTICES_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def infer_tax_year(
    date_received: Optional[str],
    explicit_tax_year: Optional[int],
) -> int:
    if explicit_tax_year is not None:
        return explicit_tax_year
    if date_received:
        return datetime.strptime(date_received, "%Y-%m-%d").year
    return datetime.now().year


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Record a tax authority notice into local tax memory."
    )
    parser.add_argument(
        "--authority",
        required=True,
        help="Authority name, e.g. IRS, California FTB, local tax office",
    )
    parser.add_argument(
        "--notice-type",
        default="letter",
        help="Notice type, e.g. letter, penalty_notice, verification_request",
    )
    parser.add_argument(
        "--date-received",
        default=datetime.now().strftime("%Y-%m-%d"),
        help="Date received in YYYY-MM-DD format",
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        default=None,
        help="Related tax year if known",
    )
    parser.add_argument(
        "--summary",
        required=True,
        help="Short summary of the notice",
    )
    parser.add_argument(
        "--response-deadline",
        default=None,
        help="Optional response deadline in YYYY-MM-DD format",
    )
    parser.add_argument(
        "--status",
        default="open",
        choices=["open", "under_review", "responded", "closed"],
        help="Current notice status",
    )
    parser.add_argument(
        "--professional-review-needed",
        action="store_true",
        help="Flag this notice for professional review",
    )

    args = parser.parse_args()

    tax_year = infer_tax_year(args.date_received, args.tax_year)
    now_iso = datetime.now().isoformat(timespec="seconds")
    notice_id = f"NOT-{str(uuid.uuid4())[:8].upper()}"

    notice = {
        "id": notice_id,
        "tax_year": tax_year,
        "authority": args.authority,
        "notice_type": args.notice_type,
        "date_received": args.date_received,
        "summary": args.summary,
        "response_deadline": args.response_deadline,
        "status": args.status,
        "professional_review_needed": args.professional_review_needed,
        "created_at": now_iso,
    }

    data = load_notices()
    data["notices"].append(notice)
    save_notices(data)

    print(f"Recorded notice: {notice_id}")
    print(f"  Tax year: {tax_year}")
    print(f"  Authority: {args.authority}")
    print(f"  Type: {args.notice_type}")
    print(f"  Date received: {args.date_received}")
    print(f"  Status: {args.status}")
    if args.response_deadline:
        print(f"  Response deadline: {args.response_deadline}")
    print(
        f"  Professional review needed: "
        f"{'yes' if args.professional_review_needed else 'no'}"
    )
    print(f"  Saved to: {NOTICES_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/prep_meeting.py

```python
#!/usr/bin/env python3
"""Generate a CPA-ready tax handoff summary in Markdown."""

import argparse
import json
import os
from datetime import datetime
from typing import Any, Dict, List


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")
EXPENSES_FILE = os.path.join(BASE_DIR, "expenses.json")
NOTICES_FILE = os.path.join(BASE_DIR, "notices.json")
QUESTIONS_FILE = os.path.join(BASE_DIR, "questions_for_cpa.json")
SUMMARIES_DIR = os.path.join(BASE_DIR, "summaries")


def ensure_dirs() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)
    os.makedirs(SUMMARIES_DIR, exist_ok=True)


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def filter_by_tax_year(items: List[Dict[str, Any]], tax_year: int) -> List[Dict[str, Any]]:
    return [item for item in items if item.get("tax_year") == tax_year]


def format_money(amount: Any, currency: str = "USD") -> str:
    if amount is None:
        return "Unknown"
    try:
        return f"{currency} {float(amount):,.2f}"
    except (ValueError, TypeError):
        return str(amount)


def summarize_expenses(expenses: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
    summary: Dict[str, Dict[str, Any]] = {}

    for expense in expenses:
        category = expense.get("category", "uncategorized")
        amount = float(expense.get("amount", 0.0))
        currency = expense.get("currency", "USD")

        if category not in summary:
            summary[category] = {
                "count": 0,
                "total": 0.0,
                "currency": currency,
            }

        summary[category]["count"] += 1
        summary[category]["total"] += amount

    return summary


def build_markdown(
    tax_year: int,
    documents: List[Dict[str, Any]],
    expenses: List[Dict[str, Any]],
    notices: List[Dict[str, Any]],
    questions: List[Dict[str, Any]],
) -> str:
    expense_summary = summarize_expenses(expenses)
    generated_at = datetime.now().isoformat(timespec="seconds")

    lines: List[str] = []
    lines.append(f"# Tax Year {tax_year} CPA Handoff Summary")
    lines.append("")
    lines.append(f"Generated at: {generated_at}")
    lines.append("")

    lines.append("## Filing Snapshot")
    lines.append("")
    lines.append(f"- Tax year: {tax_year}")
    lines.append(f"- Documents recorded: {len(documents)}")
    lines.append(f"- Expense records: {len(expenses)}")
    lines.append(f"- Open notices: {len([n for n in notices if n.get('status') != 'closed'])}")
    lines.append(f"- Open CPA questions: {len([q for q in questions if q.get('status') != 'resolved'])}")
    lines.append("")

    lines.append("## Income and Tax Documents")
    lines.append("")
    if documents:
        lines.append("| Type | Issuer | Amount | Date Received | Status |")
        lines.append("|------|--------|--------|---------------|--------|")
        for doc in documents:
            lines.append(
                f"| {doc.get('document_type', '')} | "
                f"{doc.get('issuer', '')} | "
                f"{format_money(doc.get('amount'))} | "
                f"{doc.get('date_received', '')} | "
                f"{doc.get('status', '')} |"
            )
    else:
        lines.append("_No documents recorded for this tax year._")
    lines.append("")

    lines.append("## Tax-Relevant Expense Summary")
    lines.append("")
    if expense_summary:
        lines.append("| Category | Count | Total |")
        lines.append("|----------|-------|-------|")
        for category in sorted(expense_summary.keys()):
            item = expense_summary[category]
            lines.append(
                f"| {category} | {item['count']} | "
                f"{format_money(item['total'], item['currency'])} |"
            )
    else:
        lines.append("_No expenses recorded for this tax year._")
    lines.append("")

    lines.append("## Notices")
    lines.append("")
    if notices:
        lines.append("| Authority | Type | Date Received | Response Deadline | Status | Summary |")
        lines.append("|-----------|------|---------------|-------------------|--------|---------|")
        for notice in notices:
            lines.append(
                f"| {notice.get('authority', '')} | "
                f"{notice.get('notice_type', '')} | "
                f"{notice.get('date_received', '')} | "
                f"{notice.get('response_deadline') or ''} | "
                f"{notice.get('status', '')} | "
                f"{notice.get('summary', '')} |"
            )
    else:
        lines.append("_No notices recorded for this tax year._")
    lines.append("")

    lines.append("## Questions for My CPA")
    lines.append("")
    if questions:
        unresolved = [q for q in questions if q.get("status") != "resolved"]
        if unresolved:
            for idx, question in enumerate(unresolved, start=1):
                lines.append(f"{idx}. {question.get('question', '')}")
        else:
            lines.append("_No open CPA questions for this tax year._")
    else:
        lines.append("_No CPA questions recorded for this tax year._")
    lines.append("")

    lines.append("## Notes")
    lines.append("")
    lines.append(
        "This summary is for organization and professional handoff only. "
        "It does not provide tax advice, filing positions, or legal interpretations."
    )
    lines.append("")

    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Generate a CPA-ready tax handoff summary."
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        required=True,
        help="Tax year to summarize",
    )

    args = parser.parse_args()
    tax_year = args.tax_year

    ensure_dirs()

    documents_data = load_json(DOCUMENTS_FILE, {"documents": []})
    expenses_data = load_json(EXPENSES_FILE, {"expenses": []})
    notices_data = load_json(NOTICES_FILE, {"notices": []})
    questions_data = load_json(QUESTIONS_FILE, {"questions": []})

    documents = filter_by_tax_year(documents_data.get("documents", []), tax_year)
    expenses = filter_by_tax_year(expenses_data.get("expenses", []), tax_year)
    notices = filter_by_tax_year(notices_data.get("notices", []), tax_year)
    questions = filter_by_tax_year(questions_data.get("questions", []), tax_year)

    markdown = build_markdown(tax_year, documents, expenses, notices, questions)

    output_path = os.path.join(SUMMARIES_DIR, f"summary_{tax_year}.md")
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(markdown)

    print(f"Generated CPA handoff summary for tax year {tax_year}")
    print(f"  Documents included: {len(documents)}")
    print(f"  Expenses included: {len(expenses)}")
    print(f"  Notices included: {len(notices)}")
    print(f"  Questions included: {len(questions)}")
    print(f"  Saved to: {output_path}")


if __name__ == "__main__":
    main()

```

### scripts/set_year_state.py

```python
#!/usr/bin/env python3
"""Set and update annual tax workflow state."""

import argparse
import json
import os
from datetime import datetime
from typing import Any, Dict, List


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
YEAR_STATE_FILE = os.path.join(BASE_DIR, "year_state.json")
DOCUMENTS_FILE = os.path.join(BASE_DIR, "documents.json")
EXPENSES_FILE = os.path.join(BASE_DIR, "expenses.json")
NOTICES_FILE = os.path.join(BASE_DIR, "notices.json")
QUESTIONS_FILE = os.path.join(BASE_DIR, "questions_for_cpa.json")


VALID_STATES = {
    "capturing",
    "reconciling",
    "pre_filing",
    "filed",
    "archived",
    "notice_followup",
}


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_json(path: str, default: Dict[str, Any]) -> Dict[str, Any]:
    if os.path.exists(path):
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return default


def save_json(path: str, data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def filter_by_tax_year(items: List[Dict[str, Any]], tax_year: int) -> List[Dict[str, Any]]:
    return [item for item in items if item.get("tax_year") == tax_year]


def count_open_notices(notices: List[Dict[str, Any]]) -> int:
    return len([n for n in notices if n.get("status") != "closed"])


def count_open_questions(questions: List[Dict[str, Any]]) -> int:
    return len([q for q in questions if q.get("status") != "resolved"])


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Set and update annual tax workflow state."
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        required=True,
        help="Tax year to update",
    )
    parser.add_argument(
        "--state",
        required=True,
        choices=sorted(VALID_STATES),
        help="New workflow state",
    )

    args = parser.parse_args()

    tax_year = args.tax_year
    state = args.state
    now_iso = datetime.now().isoformat(timespec="seconds")

    documents_data = load_json(DOCUMENTS_FILE, {"documents": []})
    expenses_data = load_json(EXPENSES_FILE, {"expenses": []})
    notices_data = load_json(NOTICES_FILE, {"notices": []})
    questions_data = load_json(QUESTIONS_FILE, {"questions": []})
    year_state_data = load_json(YEAR_STATE_FILE, {"years": {}})

    documents = filter_by_tax_year(documents_data.get("documents", []), tax_year)
    expenses = filter_by_tax_year(expenses_data.get("expenses", []), tax_year)
    notices = filter_by_tax_year(notices_data.get("notices", []), tax_year)
    questions = filter_by_tax_year(questions_data.get("questions", []), tax_year)

    year_state_data["years"][str(tax_year)] = {
        "state": state,
        "documents_received_count": len(documents),
        "expense_records_count": len(expenses),
        "open_notices_count": count_open_notices(notices),
        "open_questions_for_cpa_count": count_open_questions(questions),
        "updated_at": now_iso,
    }

    save_json(YEAR_STATE_FILE, year_state_data)

    print(f"Updated tax year state: {tax_year}")
    print(f"  State: {state}")
    print(f"  Documents recorded: {len(documents)}")
    print(f"  Expense records: {len(expenses)}")
    print(f"  Open notices: {count_open_notices(notices)}")
    print(f"  Open CPA questions: {count_open_questions(questions)}")
    print(f"  Saved to: {YEAR_STATE_FILE}")


if __name__ == "__main__":
    main()

```

### scripts/track_expense.py

```python
#!/usr/bin/env python3
"""Record a tax-relevant expense or receipt into local tax memory."""

import argparse
import json
import os
import uuid
from datetime import datetime
from typing import Any, Dict, Optional


BASE_DIR = os.path.expanduser("~/.openclaw/workspace/memory/tax")
EXPENSES_FILE = os.path.join(BASE_DIR, "expenses.json")


def ensure_base_dir() -> None:
    os.makedirs(BASE_DIR, exist_ok=True)


def load_expenses() -> Dict[str, Any]:
    if os.path.exists(EXPENSES_FILE):
        with open(EXPENSES_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"expenses": []}


def save_expenses(data: Dict[str, Any]) -> None:
    ensure_base_dir()
    with open(EXPENSES_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)


def infer_tax_year(date_str: Optional[str], explicit_tax_year: Optional[int]) -> int:
    if explicit_tax_year is not None:
        return explicit_tax_year
    if date_str:
        return datetime.strptime(date_str, "%Y-%m-%d").year
    return datetime.now().year


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Record a tax-relevant expense or receipt."
    )
    parser.add_argument(
        "--amount",
        type=float,
        required=True,
        help="Expense amount",
    )
    parser.add_argument(
        "--category",
        required=True,
        help="Expense category, e.g. software, business_meal, travel, advertising",
    )
    parser.add_argument(
        "--date",
        default=datetime.now().strftime("%Y-%m-%d"),
        help="Expense date in YYYY-MM-DD format",
    )
    parser.add_argument(
        "--tax-year",
        type=int,
        default=None,
        help="Tax year the record belongs to",
    )
    parser.add_argument(
        "--merchant",
        default="Unknown",
        help="Merchant or payee name",
    )
    parser.add_argument(
        "--purpose",
        default="",
        help="Optional business or recordkeeping purpose",
    )
    parser.add_argument(
        "--currency",
        default="USD",
        help="Currency code, default USD",
    )
    parser.add_argument(
        "--documentation-status",
        default="receipt_not_confirmed",
        choices=[
            "receipt_not_confirmed",
            "receipt_saved",
            "invoice_saved",
            "statement_only",
            "other_supporting_record",
        ],
        help="Documentation status for the record",
    )
    parser.add_argument(
        "--professional-review-status",
        default="unreviewed",
        choices=["unreviewed", "flagged_for_cpa", "reviewed"],
        help="Professional review status",
    )
    parser.add_argument(
        "--raw-text",
        default="",
        help="Original user wording or source text",
    )

    args = parser.parse_args()

    tax_year = infer_tax_year(args.date, args.tax_year)
    now_iso = datetime.now().isoformat(timespec="seconds")
    expense_id = f"EXP-{str(uuid.uuid4())[:8].upper()}"

    expense = {
        "id": expense_id,
        "tax_year": tax_year,
        "date": args.date,
        "amount": args.amount,
        "currency": args.currency,
        "category": args.category,
        "merchant": args.merchant,
        "purpose": args.purpose,
        "documentation_status": args.documentation_status,
        "professional_review_status": args.professional_review_status,
        "raw_text": args.raw_text,
        "created_at": now_iso,
    }

    data = load_expenses()
    data["expenses"].append(expense)
    save_expenses(data)

    print(f"Recorded expense: {expense_id}")
    print(f"  Tax year: {tax_year}")
    print(f"  Date: {args.date}")
    print(f"  Amount: {args.currency} {args.amount:,.2f}")
    print(f"  Category: {args.category}")
    print(f"  Merchant: {args.merchant}")
    if args.purpose:
        print(f"  Purpose: {args.purpose}")
    print(f"  Documentation status: {args.documentation_status}")
    print(f"  Professional review status: {args.professional_review_status}")
    print(f"  Saved to: {EXPENSES_FILE}")


if __name__ == "__main__":
    main()

```