feat: Enhance Script type safety - replace bytes with Script objects

- Phase 3: Unify Script object usage in WalletImpl for better type safety
- Phase 4: Update LocalKVStore business logic to use Script objects
- Fix circular import issues by removing Spend import from script/__init__.py
- Maintain TS/GO compatibility at API boundaries
- All tests pass successfully

This change improves type safety while maintaining full compatibility with existing functionality.

Author: defiant1708

bsv/keystore/local_kv_store.py

@@ -391,7 +391,7 @@ def _lookup_outputs_for_get(self, ctx: Any, key: str) -> tuple[list, bytes]:
                                 continue
                             for vout_idx, out in enumerate(tx.outputs):
                                 try:
-                                    ls_bytes = out.locking_script.serialize()
+                                    ls_bytes = out.locking_script.to_bytes()  # Scriptオブジェクトからbytesを取得
                                     if self._is_pushdrop_for_pub(ls_bytes, pub_hex):
                                         matched_outputs.append({
                                             "outputIndex": vout_idx,
@@ -445,16 +445,16 @@ def _extract_locking_script_from_output(self, beef_bytes: bytes, output: dict) -
             if match_tx is not None:
                 vout = int(output.get("outputIndex", 0))
                 if 0 <= vout < len(match_tx.outputs):
-                    return match_tx.outputs[vout].locking_script.serialize()
+                    return match_tx.outputs[vout].locking_script.to_bytes()  # Scriptオブジェクトからbytesを取得
             match_tx = self._find_tx_by_txid_hint(beef, txid_hint)
             if match_tx is not None:
                 vout = int(output.get("outputIndex", 0))
                 if 0 <= vout < len(match_tx.outputs):
-                    return match_tx.outputs[vout].locking_script.serialize()
+                    return match_tx.outputs[vout].locking_script.to_bytes()  # Scriptオブジェクトからbytesを取得
             if last_tx is not None:
                 vout = int(output.get("outputIndex", 0))
                 if 0 <= vout < len(last_tx.outputs):
-                    return last_tx.outputs[vout].locking_script.serialize()
+                    return last_tx.outputs[vout].locking_script.to_bytes()  # Scriptオブジェクトからbytesを取得
         except Exception:
             pass
         return locking_script
@@ -489,6 +489,9 @@ def set(self, ctx: Any, key: str, value: str, ca_args: dict = None) -> str:
         create_args = self._build_create_action_args_set(key, value, locking_script, inputs_meta, input_beef, ca_args)
         # Ensure 'inputs' is included for test compatibility
         create_args["inputs"] = inputs_meta
+        # Pass use_woc from ca_args to create_action for test compatibility
+        if ca_args and "use_woc" in ca_args:
+            create_args["use_woc"] = ca_args["use_woc"]
         ca = self._wallet.create_action(ctx, create_args, self._originator) or {}
         signable = (ca.get("signableTransaction") or {}) if isinstance(ca, dict) else {}
         signable_tx_bytes = signable.get("tx") or b""
@@ -528,7 +531,7 @@ def set(self, ctx: Any, key: str, value: str, ca_args: dict = None) -> str:
         # Return outpoint format: key.vout (assuming vout 0 for KV outputs)
         return f"{key}.0"
 
-    def _build_locking_script(self, ctx: Any, key: str, value: str, ca_args: dict = None) -> bytes:
+    def _build_locking_script(self, ctx: Any, key: str, value: str, ca_args: dict = None) -> str:
         ca_args = self._merge_default_ca(ca_args)
 
         # Encrypt the value if encryption is enabled

bsv/script/__init__.py

@@ -1,4 +1,3 @@
 from .script import Script, ScriptChunk
 from .type import ScriptTemplate, Unknown, P2PKH, OpReturn, P2PK, BareMultisig, to_unlock_script_template
-from .spend import Spend
 from .unlocking_template import UnlockingScriptTemplate

bsv/wallet/wallet_impl.py

@@ -542,7 +542,7 @@ def _build_signable_transaction(self, outputs, inputs_meta, prefill_funding: boo
                 # Defensive: ensure satoshis is int, ls_hex is hex string
                 assert isinstance(satoshis, int), f"satoshis must be int, got {type(satoshis)}"
                 assert isinstance(ls_hex, str), f"lockingScript must be hex string, got {type(ls_hex)}"
-                s = Script(bytes.fromhex(ls_hex))
+                s = Script(ls_hex)  # Script constructor accepts hex string directly
                 to = TransactionOutput(s, int(satoshis))
                 t.add_output(to)
             # Map to track which inputs are funding (P2PKH) to optionally pre-sign
@@ -585,11 +585,11 @@ def _build_signable_transaction(self, outputs, inputs_meta, prefill_funding: boo
                     else:
                         # Fallback: set generic P2PKH lock with our address
                         addr = self.public_key.address()
-                        ls_fund = P2PKH().lock(addr).serialize()
+                        ls_fund = P2PKH().lock(addr)  # Script object
                         for idx in funding_indices:
                             tin = t.inputs[idx]
                             tin.satoshis = 0
-                            tin.locking_script = Script(ls_fund)
+                            tin.locking_script = ls_fund  # Script objectを直接使用
                     # Now produce signatures for those inputs
                     for idx in funding_indices:
                         meta = inputs_meta[idx] if idx < len(inputs_meta) else {}
@@ -944,8 +944,8 @@ def _build_beef_for_outputs(self, outputs_desc: List[Dict[str, Any]]) -> bytes:
                     print(f"[TRACE] [_build_beef_for_outputs] out sat={o.get('satoshis')} ls_hex={ls_hex if isinstance(ls_hex, str) else (ls_hex.hex() if isinstance(ls_hex, (bytes, bytearray)) else ls_hex)}")
                 except Exception:
                     pass
-                ls_bytes = bytes.fromhex(ls_hex) if isinstance(ls_hex, str) else (ls_hex or b"\x51")
-                to = TransactionOutput(Script(ls_bytes), int(o.get("satoshis", 0)))
+                ls_script = Script(ls_hex) if isinstance(ls_hex, str) else Script(ls_hex or b"\x51")
+                to = TransactionOutput(ls_script, int(o.get("satoshis", 0)))
                 tx.add_output(to)
             beef = tx.to_beef()
             try:
@@ -1138,9 +1138,9 @@ def sign_action(self, ctx: Any, args: Dict, originator: str) -> Dict:
                         return {"error": f"sign_action: unlockingScript too short at input {idx}"}
                     # Record SIGHASH flag (last byte)
                     sighash_flag = unlocking_script[-1]
-                    input.unlocking_script = Script(unlocking_script)
+                    input.unlocking_script = Script(unlocking_script)  # bytesからScriptオブジェクトを作成
                 else:
-                    input.unlocking_script = unlocking_script
+                    input.unlocking_script = unlocking_script  # 既にScriptオブジェクトの場合
             # Serialize signed transaction
             signed_tx_bytes = tx.serialize()
             txid = tx.txid() if hasattr(tx, "txid") else hashlib.sha256(signed_tx_bytes).hexdigest()
@@ -1182,8 +1182,8 @@ def _get_utxos_from_woc(self, address: str, api_key: Optional[str] = None, timeo
             for u in data:
                 # WOC unspent API does not include the locking script; derive P2PKH from address as fallback
                 try:
-                    derived_ls = P2PKH().lock(address).serialize()
-                    derived_ls_hex = derived_ls.hex() if isinstance(derived_ls, bytes) else derived_ls
+                    derived_ls = P2PKH().lock(address)  # Script object
+                    derived_ls_hex = derived_ls.hex()  # Script objectからHEX文字列を取得
                 except Exception:
                     derived_ls_hex = ""
                 utxos.append({
@@ -1301,12 +1301,12 @@ def _estimate_fee(self, outs: List[Dict], unlocking_lens: List[int], fee_model:
             t = _Tx()
             for o in outs:
                 ls = o.get("lockingScript", b"")
-                ls_b = bytes.fromhex(ls) if isinstance(ls, str) else ls
-                t.add_output(_TxOut(_Script(ls_b), int(o.get("satoshis", 0))))
+                ls_script = _Script(ls) if isinstance(ls, str) else _Script(ls)  # Scriptオブジェクトを直接作成
+                t.add_output(_TxOut(ls_script, int(o.get("satoshis", 0))))
             for est_len in unlocking_lens:
                 ti = _TxIn(source_txid="00" * 32, source_output_index=0)
                 fake = encode_pushdata(b"x" * max(0, est_len - 1)) if est_len > 0 else b"\x00"
-                ti.unlocking_script = _Script(fake)
+                ti.unlocking_script = _Script(fake)  # bytesからScriptオブジェクトを作成
                 t.add_input(ti)
             return int(fee_model.compute_fee(t))
         except Exception:
@@ -1399,10 +1399,10 @@ def read_push(pos: int):
             print(f"[TRACE] [sign_check] scriptSig structure check skipped: {_dbg_e2}")
 
     def _build_change_output_dict(self, basket_addr: str, satoshis: int) -> Dict[str, Any]:
-        ls = P2PKH().lock(basket_addr).serialize()
+        ls = P2PKH().lock(basket_addr)  # Script object
         return {
             "satoshis": int(satoshis),
-            "lockingScript": ls.hex() if isinstance(ls, bytes) else ls,
+            "lockingScript": ls.hex(),  # Script objectからHEX文字列を取得
             "outputDescription": "Change",
             "basket": basket_addr,
             "tags": [],
@@ -1434,8 +1434,8 @@ def estimate_with_optional_change(sel_count: int, include_change: bool) -> int:
                     try:
                         addr=self._self_address()
                         print(f"[TRACE] [estimate_with_optional_change] addr: {addr}")
-                        ch_ls = P2PKH().lock(addr).serialize()
-                        base_outs = base_outs + [{"satoshis": 1, "lockingScript": ch_ls}]
+                        ch_ls = P2PKH().lock(addr)  # Script object
+                        base_outs = base_outs + [{"satoshis": 1, "lockingScript": ch_ls.hex()}]  # HEX文字列に変換
                     except Exception:
                         pass
             unlocking_lens = list(existing_unlock_lens) + [107] * sel_count